// processPeerUpdate process a peer update message. For new or reconnected peers,
// we create a peer state if one does not exist for the peer, which should always
// be the case, and we spawn all the relevant goroutine to broadcast messages to
// the peer. During peer removal, we remove the peer for our set of peers and
// signal to all spawned goroutines to gracefully exit in a non-blocking manner.
processPeerUpdate process a peer update message. For new or reconnected peers,
we create a peer state if one does not exist for the peer, which should always
be the case, and we spawn all the relevant goroutine to broadcast messages to
the peer. During peer removal, we remove the peer for our set of peers and
signal to all spawned goroutines to gracefully exit in a non-blocking manner.
// processPeerUpdate process a peer update message . For new or reconnected peers // we create a peer state if one does not exist for the peer which should always // be the case and we spawn all the relevant goroutine to broadcast messages to // the peer . During peer removal we remove the peer for our set of peers and // signal to all spawned goroutines to gracefully exit in a non - blocking manner .
func (r *Reactor) processPeerUpdate(peerUpdate p2p.PeerUpdate) {
	r.Logger.Debug("received peer update", "peer", peerUpdate.NodeID, "status", peerUpdate.Status)
	r.mtx.Lock()
	defer r.mtx.Unlock()
	switch peerUpdate.Status {
	case p2p.PeerStatusUp:
		if !r.IsRunning() {
			return
		}
		var (
			ps *PeerState
			ok bool
		)
		ps, ok = r.peers[peerUpdate.NodeID]
		if !ok {
			ps = NewPeerState(r.Logger, peerUpdate.NodeID)
			r.peers[peerUpdate.NodeID] = ps
		}
		if !ps.IsRunning() {
			ps.broadcastWG.Add(3)
			ps.SetRunning(true)
			go r.gossipDataRoutine(ps)
			go r.gossipVotesRoutine(ps)
			go r.queryMaj23Routine(ps)
			if !r.waitSync {
				go r.sendNewRoundStepMessage(ps.peerID)
			}
		}
	case p2p.PeerStatusDown:
		ps, ok := r.peers[peerUpdate.NodeID]
		if ok && ps.IsRunning() {
			ps.closer.Close()
			go func() {
				ps.broadcastWG.Wait()
				r.mtx.Lock()
				delete(r.peers, peerUpdate.NodeID)
				r.mtx.Unlock()
				ps.SetRunning(false)
			}()
		}
	}
}
func ( r * Reactor ) processPeerUpdate ( peerUpdate p2p . PeerUpdate ) { r . Logger . Debug ( " " , " " , peerUpdate . NodeID , " " , peerUpdate . Status ) r . mtx . Lock ( ) defer r . mtx . Unlock ( ) switch peerUpdate . Status { case p2p . PeerStatusUp : if ! r . IsRunning ( ) { return } var ( ps * PeerState ok bool ) ps , ok = r . peers [ peerUpdate . NodeID ] if ! ok { ps = NewPeerState ( r . Logger , peerUpdate . NodeID ) r . peers [ peerUpdate . NodeID ] = ps } if ! ps . IsRunning ( ) { ps . broadcastWG . Add ( 3 ) ps . SetRunning ( true ) go r . gossipDataRoutine ( ps ) go r . gossipVotesRoutine ( ps ) go r . queryMaj23Routine ( ps ) if ! r . waitSync { go r . sendNewRoundStepMessage ( ps . peerID ) } } case p2p . PeerStatusDown : ps , ok := r . peers [ peerUpdate . NodeID ] if ok && ps . IsRunning ( ) { ps . closer . Close ( ) go func ( ) { ps . broadcastWG . Wait ( ) r . mtx . Lock ( ) delete ( r . peers , peerUpdate . NodeID ) r . mtx . Unlock ( ) ps . SetRunning ( false ) } ( ) } } }
processPeerUpdate process a peer update message.
processPeerUpdate process a peer update message .
// Do not allow starting new broadcasting goroutines after reactor shutdown // has been initiated. This can happen after we've manually closed all // peer goroutines and closed r.closeCh, but the router still sends in-flight // peer updates. // Set the peer state's closer to signal to all spawned goroutines to exit // 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. // start goroutines for this peer // Send our state to the peer. If we're block-syncing, broadcast a // RoundStepMessage later upon SwitchToConsensus(). // signal to all spawned goroutines for the peer to gracefully exit // Wait for all spawned broadcast goroutines to exit before marking the // peer state as no longer running and removal from the peers map.