From 710407e9b2d50078aa5bf38b4ce0aa6c27e74a18 Mon Sep 17 00:00:00 2001 From: Sam Kleinman Date: Fri, 4 Feb 2022 10:18:19 -0500 Subject: [PATCH] consensus: delay start of peer routines (backport of #7753) (#7760) --- internal/consensus/peer_state.go | 4 +- internal/consensus/reactor.go | 185 +++++++++++++++++------------ internal/statesync/reactor_test.go | 3 + 3 files changed, 110 insertions(+), 82 deletions(-) diff --git a/internal/consensus/peer_state.go b/internal/consensus/peer_state.go index 6a64e8e10..9c092056b 100644 --- a/internal/consensus/peer_state.go +++ b/internal/consensus/peer_state.go @@ -3,7 +3,6 @@ package consensus import ( "errors" "fmt" - "sync" "time" cstypes "github.com/tendermint/tendermint/internal/consensus/types" @@ -45,8 +44,7 @@ type PeerState struct { PRS cstypes.PeerRoundState `json:"round_state"` Stats *peerStateStats `json:"stats"` - broadcastWG sync.WaitGroup - closer *tmsync.Closer + closer *tmsync.Closer } // NewPeerState returns a new PeerState for the given node ID. diff --git a/internal/consensus/reactor.go b/internal/consensus/reactor.go index 22bd7576e..214b41430 100644 --- a/internal/consensus/reactor.go +++ b/internal/consensus/reactor.go @@ -3,10 +3,10 @@ package consensus import ( "fmt" "runtime/debug" + "sync" "time" cstypes "github.com/tendermint/tendermint/internal/consensus/types" - tmsync "github.com/tendermint/tendermint/internal/libs/sync" "github.com/tendermint/tendermint/internal/p2p" sm "github.com/tendermint/tendermint/internal/state" "github.com/tendermint/tendermint/libs/bits" @@ -128,9 +128,10 @@ type Reactor struct { eventBus *types.EventBus Metrics *Metrics - mtx tmsync.RWMutex - peers map[types.NodeID]*PeerState - waitSync bool + mtx sync.RWMutex + peers map[types.NodeID]*PeerState + waitSync bool + readySignal chan struct{} // closed when the node is ready to start consensus stateCh *p2p.Channel dataCh *p2p.Channel @@ -138,12 +139,7 @@ type Reactor struct { voteSetBitsCh *p2p.Channel peerUpdates *p2p.PeerUpdates - // NOTE: We need a dedicated stateCloseCh channel for signaling closure of - // the StateChannel due to the fact that the StateChannel message handler - // performs a send on the VoteSetBitsChannel. This is an antipattern, so having - // this dedicated channel,stateCloseCh, is necessary in order to avoid data races. - stateCloseCh chan struct{} - closeCh chan struct{} + closeCh chan struct{} } // NewReactor returns a reference to a new consensus reactor, which implements @@ -172,7 +168,7 @@ func NewReactor( voteCh: voteCh, voteSetBitsCh: voteSetBitsCh, peerUpdates: peerUpdates, - stateCloseCh: make(chan struct{}), + readySignal: make(chan struct{}), closeCh: make(chan struct{}), } r.BaseService = *service.NewBaseService(logger, "Consensus", r) @@ -181,6 +177,10 @@ func NewReactor( opt(r) } + if !r.waitSync { + close(r.readySignal) + } + return r } @@ -235,26 +235,13 @@ func (r *Reactor) OnStop() { // lock to complete any of the methods that the waitgroup is waiting on. for _, state := range r.peers { state.closer.Close() - state.broadcastWG.Wait() } r.mtx.Unlock() - // Close the StateChannel goroutine separately since it uses its own channel - // to signal closure. - close(r.stateCloseCh) - <-r.stateCh.Done() - // Close closeCh to signal to all spawned goroutines to gracefully exit. All // p2p Channels should execute Close(). close(r.closeCh) - // Wait for all p2p Channels to be closed before returning. This ensures we - // can easily reason about synchronization of all p2p Channels and ensure no - // panics will occur. - <-r.voteSetBitsCh.Done() - <-r.dataCh.Done() - <-r.voteCh.Done() - <-r.peerUpdates.Done() } // SetEventBus sets the reactor's event bus. @@ -292,6 +279,7 @@ func (r *Reactor) SwitchToConsensus(state sm.State, skipWAL bool) { r.mtx.Lock() r.waitSync = false + close(r.readySignal) r.mtx.Unlock() r.Metrics.BlockSyncing.Set(0) @@ -353,15 +341,23 @@ func (r *Reactor) GetPeerState(peerID types.NodeID) (*PeerState, bool) { } func (r *Reactor) broadcastNewRoundStepMessage(rs *cstypes.RoundState) { - r.stateCh.Out <- p2p.Envelope{ + select { + case r.stateCh.Out <- p2p.Envelope{ Broadcast: true, Message: makeRoundStepMessage(rs), + }: + case <-r.closeCh: + return + } } func (r *Reactor) broadcastNewValidBlockMessage(rs *cstypes.RoundState) { psHeader := rs.ProposalBlockParts.Header() - r.stateCh.Out <- p2p.Envelope{ + select { + case <-r.closeCh: + return + case r.stateCh.Out <- p2p.Envelope{ Broadcast: true, Message: &tmcons.NewValidBlock{ Height: rs.Height, @@ -370,11 +366,15 @@ func (r *Reactor) broadcastNewValidBlockMessage(rs *cstypes.RoundState) { BlockParts: rs.ProposalBlockParts.BitArray().ToProto(), IsCommit: rs.Step == cstypes.RoundStepCommit, }, + }: } } func (r *Reactor) broadcastHasVoteMessage(vote *types.Vote) { - r.stateCh.Out <- p2p.Envelope{ + select { + case <-r.closeCh: + return + case r.stateCh.Out <- p2p.Envelope{ Broadcast: true, Message: &tmcons.HasVote{ Height: vote.Height, @@ -382,6 +382,7 @@ func (r *Reactor) broadcastHasVoteMessage(vote *types.Vote) { Type: vote.Type, Index: vote.ValidatorIndex, }, + }: } } @@ -444,9 +445,13 @@ func makeRoundStepMessage(rs *cstypes.RoundState) *tmcons.NewRoundStep { func (r *Reactor) sendNewRoundStepMessage(peerID types.NodeID) { rs := r.state.GetRoundState() msg := makeRoundStepMessage(rs) - r.stateCh.Out <- p2p.Envelope{ + select { + case <-r.closeCh: + return + case r.stateCh.Out <- p2p.Envelope{ To: peerID, Message: msg, + }: } } @@ -517,8 +522,6 @@ func (r *Reactor) gossipDataForCatchup(rs *cstypes.RoundState, prs *cstypes.Peer func (r *Reactor) gossipDataRoutine(ps *PeerState) { logger := r.Logger.With("peer", ps.peerID) - defer ps.broadcastWG.Done() - OUTER_LOOP: for { if !r.IsRunning() { @@ -526,6 +529,8 @@ OUTER_LOOP: } select { + case <-r.closeCh: + return case <-ps.closer.Done(): // The peer is marked for removal via a PeerUpdate as the doneCh was // explicitly closed to signal we should exit. @@ -609,11 +614,16 @@ OUTER_LOOP: propProto := rs.Proposal.ToProto() logger.Debug("sending proposal", "height", prs.Height, "round", prs.Round) - r.dataCh.Out <- p2p.Envelope{ + select { + case <-r.closeCh: + return + case r.dataCh.Out <- p2p.Envelope{ To: ps.peerID, Message: &tmcons.Proposal{ Proposal: *propProto, }, + }: + } // NOTE: A peer might have received a different proposal message, so @@ -630,13 +640,17 @@ OUTER_LOOP: pPolProto := pPol.ToProto() logger.Debug("sending POL", "height", prs.Height, "round", prs.Round) - r.dataCh.Out <- p2p.Envelope{ + select { + case <-r.closeCh: + return + case r.dataCh.Out <- p2p.Envelope{ To: ps.peerID, Message: &tmcons.ProposalPOL{ Height: rs.Height, ProposalPolRound: rs.Proposal.POLRound, ProposalPol: *pPolProto, }, + }: } } @@ -654,11 +668,16 @@ OUTER_LOOP: func (r *Reactor) pickSendVote(ps *PeerState, votes types.VoteSetReader) bool { if vote, ok := ps.PickVoteToSend(votes); ok { r.Logger.Debug("sending vote message", "ps", ps, "vote", vote) - r.voteCh.Out <- p2p.Envelope{ + select { + case <-r.closeCh: + return false + case r.voteCh.Out <- p2p.Envelope{ To: ps.peerID, Message: &tmcons.Vote{ Vote: vote.ToProto(), }, + }: + } ps.SetHasVote(vote) @@ -729,8 +748,6 @@ func (r *Reactor) gossipVotesForHeight(rs *cstypes.RoundState, prs *cstypes.Peer func (r *Reactor) gossipVotesRoutine(ps *PeerState) { logger := r.Logger.With("peer", ps.peerID) - defer ps.broadcastWG.Done() - // XXX: simple hack to throttle logs upon sleep logThrottle := 0 @@ -809,8 +826,6 @@ OUTER_LOOP: // NOTE: `queryMaj23Routine` has a simple crude design since it only comes // into play for liveness when there's a signature DDoS attack happening. func (r *Reactor) queryMaj23Routine(ps *PeerState) { - defer ps.broadcastWG.Done() - OUTER_LOOP: for { if !r.IsRunning() { @@ -818,6 +833,8 @@ OUTER_LOOP: } select { + case <-r.closeCh: + return case <-ps.closer.Done(): // The peer is marked for removal via a PeerUpdate as the doneCh was // explicitly closed to signal we should exit. @@ -833,7 +850,12 @@ OUTER_LOOP: if rs.Height == prs.Height { if maj23, ok := rs.Votes.Prevotes(prs.Round).TwoThirdsMajority(); ok { - r.stateCh.Out <- p2p.Envelope{ + select { + case <-ps.closer.Done(): + return + case <-r.closeCh: + return + case r.stateCh.Out <- p2p.Envelope{ To: ps.peerID, Message: &tmcons.VoteSetMaj23{ Height: prs.Height, @@ -841,6 +863,8 @@ OUTER_LOOP: Type: tmproto.PrevoteType, BlockID: maj23.ToProto(), }, + }: + } time.Sleep(r.state.config.PeerQueryMaj23SleepDuration) @@ -855,7 +879,12 @@ OUTER_LOOP: if rs.Height == prs.Height { if maj23, ok := rs.Votes.Precommits(prs.Round).TwoThirdsMajority(); ok { - r.stateCh.Out <- p2p.Envelope{ + select { + case <-ps.closer.Done(): + return + case <-r.closeCh: + return + case r.stateCh.Out <- p2p.Envelope{ To: ps.peerID, Message: &tmcons.VoteSetMaj23{ Height: prs.Height, @@ -863,6 +892,7 @@ OUTER_LOOP: Type: tmproto.PrecommitType, BlockID: maj23.ToProto(), }, + }: } time.Sleep(r.state.config.PeerQueryMaj23SleepDuration) @@ -877,7 +907,12 @@ OUTER_LOOP: if rs.Height == prs.Height && prs.ProposalPOLRound >= 0 { if maj23, ok := rs.Votes.Prevotes(prs.ProposalPOLRound).TwoThirdsMajority(); ok { - r.stateCh.Out <- p2p.Envelope{ + select { + case <-ps.closer.Done(): + return + case <-r.closeCh: + return + case r.stateCh.Out <- p2p.Envelope{ To: ps.peerID, Message: &tmcons.VoteSetMaj23{ Height: prs.Height, @@ -885,8 +920,8 @@ OUTER_LOOP: Type: tmproto.PrevoteType, BlockID: maj23.ToProto(), }, + }: } - time.Sleep(r.state.config.PeerQueryMaj23SleepDuration) } } @@ -902,7 +937,12 @@ OUTER_LOOP: if prs.CatchupCommitRound != -1 && prs.Height > 0 && prs.Height <= r.state.blockStore.Height() && prs.Height >= r.state.blockStore.Base() { if commit := r.state.LoadCommit(prs.Height); commit != nil { - r.stateCh.Out <- p2p.Envelope{ + select { + case <-ps.closer.Done(): + return + case <-r.closeCh: + return + case r.stateCh.Out <- p2p.Envelope{ To: ps.peerID, Message: &tmcons.VoteSetMaj23{ Height: prs.Height, @@ -910,6 +950,7 @@ OUTER_LOOP: Type: tmproto.PrecommitType, BlockID: commit.BlockID.ToProto(), }, + }: } time.Sleep(r.state.config.PeerQueryMaj23SleepDuration) @@ -943,12 +984,7 @@ func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) { return } - var ( - ps *PeerState - ok bool - ) - - ps, ok = r.peers[peerUpdate.NodeID] + ps, ok := r.peers[peerUpdate.NodeID] if !ok { ps = NewPeerState(r.Logger, peerUpdate.NodeID) r.peers[peerUpdate.NodeID] = ps @@ -959,19 +995,30 @@ func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) { // when the peer is removed. We also set the running state to ensure we // do not spawn multiple instances of the same goroutines and finally we // set the waitgroup counter so we know when all goroutines have exited. - ps.broadcastWG.Add(3) ps.SetRunning(true) - // start goroutines for this peer - go r.gossipDataRoutine(ps) - go r.gossipVotesRoutine(ps) - go r.queryMaj23Routine(ps) + go func() { + select { + case <-r.closeCh: + return + case <-r.readySignal: + // do nothing if the peer has + // stopped while we've been waiting. + if !ps.IsRunning() { + return + } + // start goroutines for this peer + go r.gossipDataRoutine(ps) + go r.gossipVotesRoutine(ps) + go r.queryMaj23Routine(ps) - // Send our state to the peer. If we're block-syncing, broadcast a - // RoundStepMessage later upon SwitchToConsensus(). - if !r.waitSync { - go r.sendNewRoundStepMessage(ps.peerID) - } + // Send our state to the peer. If we're block-syncing, broadcast a + // RoundStepMessage later upon SwitchToConsensus(). + if !r.WaitSync() { + go func() { r.sendNewRoundStepMessage(ps.peerID) }() + } + } + }() } case p2p.PeerStatusDown: @@ -981,10 +1028,6 @@ func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) { ps.closer.Close() go func() { - // Wait for all spawned broadcast goroutines to exit before marking the - // peer state as no longer running and removal from the peers map. - ps.broadcastWG.Wait() - r.mtx.Lock() delete(r.peers, peerUpdate.NodeID) r.mtx.Unlock() @@ -1278,8 +1321,6 @@ func (r *Reactor) handleMessage(chID p2p.ChannelID, envelope p2p.Envelope) (err // the reactor is stopped, we will catch the signal and close the p2p Channel // gracefully. func (r *Reactor) processStateCh() { - defer r.stateCh.Close() - for { select { case envelope := <-r.stateCh.In: @@ -1290,9 +1331,7 @@ func (r *Reactor) processStateCh() { Err: err, } } - - case <-r.stateCloseCh: - r.Logger.Debug("stopped listening on StateChannel; closing...") + case <-r.closeCh: return } } @@ -1304,8 +1343,6 @@ func (r *Reactor) processStateCh() { // the reactor is stopped, we will catch the signal and close the p2p Channel // gracefully. func (r *Reactor) processDataCh() { - defer r.dataCh.Close() - for { select { case envelope := <-r.dataCh.In: @@ -1318,7 +1355,6 @@ func (r *Reactor) processDataCh() { } case <-r.closeCh: - r.Logger.Debug("stopped listening on DataChannel; closing...") return } } @@ -1330,8 +1366,6 @@ func (r *Reactor) processDataCh() { // the reactor is stopped, we will catch the signal and close the p2p Channel // gracefully. func (r *Reactor) processVoteCh() { - defer r.voteCh.Close() - for { select { case envelope := <-r.voteCh.In: @@ -1344,7 +1378,6 @@ func (r *Reactor) processVoteCh() { } case <-r.closeCh: - r.Logger.Debug("stopped listening on VoteChannel; closing...") return } } @@ -1356,8 +1389,6 @@ func (r *Reactor) processVoteCh() { // When the reactor is stopped, we will catch the signal and close the p2p // Channel gracefully. func (r *Reactor) processVoteSetBitsCh() { - defer r.voteSetBitsCh.Close() - for { select { case envelope := <-r.voteSetBitsCh.In: @@ -1370,7 +1401,6 @@ func (r *Reactor) processVoteSetBitsCh() { } case <-r.closeCh: - r.Logger.Debug("stopped listening on VoteSetBitsChannel; closing...") return } } @@ -1380,15 +1410,12 @@ func (r *Reactor) processVoteSetBitsCh() { // PeerUpdate messages. When the reactor is stopped, we will catch the signal and // close the p2p PeerUpdatesCh gracefully. func (r *Reactor) processPeerUpdates() { - defer r.peerUpdates.Close() - for { select { case peerUpdate := <-r.peerUpdates.Updates(): r.processPeerUpdate(peerUpdate) case <-r.closeCh: - r.Logger.Debug("stopped listening on peer updates channel; closing...") return } } diff --git a/internal/statesync/reactor_test.go b/internal/statesync/reactor_test.go index b90e5fd78..6e7a0dc78 100644 --- a/internal/statesync/reactor_test.go +++ b/internal/statesync/reactor_test.go @@ -181,6 +181,7 @@ func setup( t.Cleanup(func() { require.NoError(t, rts.reactor.Stop()) + rts.reactor.Wait() require.False(t, rts.reactor.IsRunning()) }) @@ -501,6 +502,8 @@ func TestReactor_BlockProviders(t *testing.T) { } func TestReactor_StateProviderP2P(t *testing.T) { + t.Cleanup(leaktest.CheckTimeout(t, 1*time.Minute)) + rts := setup(t, nil, nil, nil, 2) // make syncer non nil else test won't think we are state syncing rts.reactor.syncer = rts.syncer