mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-08-18 13:16:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

consensus: delay start of peer routines (backport of #7753) (#7760)

Browse Source
This commit is contained in:
Sam Kleinman
2022-02-04 10:18:19 -05:00
committed by GitHub
parent 5cedb588e1
commit 710407e9b2
3 changed files with 110 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/consensus/peer_state.go
+1 -3
View File
@@ -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.
internal/consensus/reactor.go
+106 -79
View File
@@ -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
}
}
internal/statesync/reactor_test.go
+3
View File
@@ -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