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

broadcast and sends updated in consensus reactor

Browse Source
This commit is contained in:
William Banfield
2022-10-19 11:39:53 -04:00
parent 833e4a15d4
commit 7f9e609720
3 changed files with 208 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
consensus/reactor.go
+143 -25
View File
@@ -292,15 +292,19 @@ func (conR *Reactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
default:
panic("Bad VoteSetBitsMessage field Type. Forgot to add a check in ValidateBasic?")
}
p, err := MsgToProto(&VoteSetBitsMessage{
Height: msg.Height,
Round: msg.Round,
Type: msg.Type,
BlockID: msg.BlockID,
Votes: ourVotes,
})
if err != nil {
panic(err)
}
src.NewTrySend(p2p.Envelope{
ChannelID: VoteSetBitsChannel,
Message: MsgToProto(&VoteSetBitsMessage{
Height: msg.Height,
Round: msg.Round,
Type: msg.Type,
BlockID: msg.BlockID,
Votes: ourVotes,
}),
Message: p,
})
default:
conR.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
@@ -433,7 +437,15 @@ func (conR *Reactor) unsubscribeFromBroadcastEvents() {
func (conR *Reactor) broadcastNewRoundStepMessage(rs *cstypes.RoundState) {
nrsMsg := makeRoundStepMessage(rs)
conR.Switch.Broadcast(StateChannel, MustEncode(nrsMsg))
p, err := MsgToProto(nrsMsg)
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: StateChannel,
Message: p,
}
conR.Switch.NewBroadcast(e)
}
func (conR *Reactor) broadcastNewValidBlockMessage(rs *cstypes.RoundState) {
@@ -444,7 +456,15 @@ func (conR *Reactor) broadcastNewValidBlockMessage(rs *cstypes.RoundState) {
BlockParts: rs.ProposalBlockParts.BitArray(),
IsCommit: rs.Step == cstypes.RoundStepCommit,
}
conR.Switch.Broadcast(StateChannel, MustEncode(csMsg))
p, err := MsgToProto(csMsg)
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: StateChannel,
Message: p,
}
conR.Switch.NewBroadcast(e)
}
// Broadcasts HasVoteMessage to peers that care.
@@ -455,7 +475,15 @@ func (conR *Reactor) broadcastHasVoteMessage(vote *types.Vote) {
Type: vote.Type,
Index: vote.ValidatorIndex,
}
conR.Switch.Broadcast(StateChannel, MustEncode(msg))
p, err := MsgToProto(msg)
if err != nil {
panic(err) // todo update the broadcast logic.
}
e := p2p.Envelope{
ChannelID: StateChannel,
Message: p,
}
conR.Switch.NewBroadcast(e)
/*
// TODO: Make this broadcast more selective.
for _, peer := range conR.Switch.Peers().List() {
@@ -466,7 +494,11 @@ func (conR *Reactor) broadcastHasVoteMessage(vote *types.Vote) {
prs := ps.GetRoundState()
if prs.Height == vote.Height {
// TODO: Also filter on round?
peer.TrySend(StateChannel, struct{ ConsensusMessage }{msg})
e := p2p.Envelope{
ChannelID: StateChannel, struct{ ConsensusMessage }{msg},
Message: p,
}
peer.NewTrySend(e)
} else {
// Height doesn't match
// TODO: check a field, maybe CatchupCommitRound?
@@ -490,7 +522,15 @@ func makeRoundStepMessage(rs *cstypes.RoundState) (nrsMsg *NewRoundStepMessage)
func (conR *Reactor) sendNewRoundStepMessage(peer p2p.Peer) {
rs := conR.getRoundState()
nrsMsg := makeRoundStepMessage(rs)
peer.Send(StateChannel, MustEncode(nrsMsg))
p, err := MsgToProto(nrsMsg)
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: StateChannel,
Message: p,
}
peer.NewSend(e)
}
func (conR *Reactor) updateRoundStateRoutine() {
@@ -535,7 +575,16 @@ OUTER_LOOP:
Part: part,
}
logger.Debug("Sending block part", "height", prs.Height, "round", prs.Round)
if peer.Send(DataChannel, MustEncode(msg)) {
p, err := MsgToProto(msg)
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: DataChannel,
Message: p,
}
if peer.NewSend(e) {
ps.SetHasProposalBlockPart(prs.Height, prs.Round, index)
}
continue OUTER_LOOP
@@ -583,7 +632,16 @@ OUTER_LOOP:
{
msg := &ProposalMessage{Proposal: rs.Proposal}
logger.Debug("Sending proposal", "height", prs.Height, "round", prs.Round)
if peer.Send(DataChannel, MustEncode(msg)) {
p, err := MsgToProto(msg)
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: DataChannel,
Message: p,
}
if peer.NewSend(e) {
// NOTE[ZM]: A peer might have received different proposal msg so this Proposal msg will be rejected!
ps.SetHasProposal(rs.Proposal)
}
@@ -599,7 +657,16 @@ OUTER_LOOP:
ProposalPOL: rs.Votes.Prevotes(rs.Proposal.POLRound).BitArray(),
}
logger.Debug("Sending POL", "height", prs.Height, "round", prs.Round)
peer.Send(DataChannel, MustEncode(msg))
p, err := MsgToProto(msg)
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: DataChannel,
Message: p,
}
peer.NewSend(e)
}
continue OUTER_LOOP
}
@@ -642,7 +709,16 @@ func (conR *Reactor) gossipDataForCatchup(logger log.Logger, rs *cstypes.RoundSt
Part: part,
}
logger.Debug("Sending block part for catchup", "round", prs.Round, "index", index)
if peer.Send(DataChannel, MustEncode(msg)) {
p, err := MsgToProto(msg)
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: DataChannel,
Message: p,
}
if peer.NewSend(e) {
ps.SetHasProposalBlockPart(prs.Height, prs.Round, index)
} else {
logger.Debug("Sending block part for catchup failed")
@@ -801,12 +877,20 @@ OUTER_LOOP:
prs := ps.GetRoundState()
if rs.Height == prs.Height {
if maj23, ok := rs.Votes.Prevotes(prs.Round).TwoThirdsMajority(); ok {
peer.TrySend(StateChannel, MustEncode(&VoteSetMaj23Message{
p, err := MsgToProto(&VoteSetMaj23Message{
Height: prs.Height,
Round: prs.Round,
Type: tmproto.PrevoteType,
BlockID: maj23,
}))
})
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: StateChannel,
Message: p,
}
peer.NewTrySend(e)
time.Sleep(conR.conS.config.PeerQueryMaj23SleepDuration)
}
}
@@ -818,12 +902,20 @@ OUTER_LOOP:
prs := ps.GetRoundState()
if rs.Height == prs.Height {
if maj23, ok := rs.Votes.Precommits(prs.Round).TwoThirdsMajority(); ok {
peer.TrySend(StateChannel, MustEncode(&VoteSetMaj23Message{
p, err := MsgToProto(&VoteSetMaj23Message{
Height: prs.Height,
Round: prs.Round,
Type: tmproto.PrecommitType,
BlockID: maj23,
}))
})
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: StateChannel,
Message: p,
}
peer.NewTrySend(e)
time.Sleep(conR.conS.config.PeerQueryMaj23SleepDuration)
}
}
@@ -835,12 +927,21 @@ OUTER_LOOP:
prs := ps.GetRoundState()
if rs.Height == prs.Height && prs.ProposalPOLRound >= 0 {
if maj23, ok := rs.Votes.Prevotes(prs.ProposalPOLRound).TwoThirdsMajority(); ok {
peer.TrySend(StateChannel, MustEncode(&VoteSetMaj23Message{
p, err := MsgToProto(&VoteSetMaj23Message{
Height: prs.Height,
Round: prs.ProposalPOLRound,
Type: tmproto.PrevoteType,
BlockID: maj23,
}))
})
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: StateChannel,
Message: p,
}
peer.NewTrySend(e)
time.Sleep(conR.conS.config.PeerQueryMaj23SleepDuration)
}
}
@@ -855,12 +956,20 @@ OUTER_LOOP:
if prs.CatchupCommitRound != -1 && prs.Height > 0 && prs.Height <= conR.conS.blockStore.Height() &&
prs.Height >= conR.conS.blockStore.Base() {
if commit := conR.conS.LoadCommit(prs.Height); commit != nil {
peer.TrySend(StateChannel, MustEncode(&VoteSetMaj23Message{
p, err := MsgToProto(&VoteSetMaj23Message{
Height: prs.Height,
Round: commit.Round,
Type: tmproto.PrecommitType,
BlockID: commit.BlockID,
}))
})
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: StateChannel,
Message: p,
}
peer.NewTrySend(e)
time.Sleep(conR.conS.config.PeerQueryMaj23SleepDuration)
}
}
@@ -1076,7 +1185,16 @@ func (ps *PeerState) PickSendVote(votes types.VoteSetReader) bool {
if vote, ok := ps.PickVoteToSend(votes); ok {
msg := &VoteMessage{vote}
ps.logger.Debug("Sending vote message", "ps", ps, "vote", vote)
if ps.peer.Send(VoteChannel, MustEncode(msg)) {
p, err := MsgToProto(msg)
if err != nil {
panic(err)
}
e := p2p.Envelope{
ChannelID: VoteChannel,
Message: p,
}
if ps.peer.NewSend(e) {
ps.SetHasVote(vote)
return true
}
p2p/peer.go
+30 -6
View File
@@ -5,10 +5,11 @@ import (
"net"
"time"
"github.com/cosmos/gogoproto/proto"
"github.com/tendermint/tendermint/libs/cmap"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/libs/service"
"google.golang.org/protobuf/proto"
tmconn "github.com/tendermint/tendermint/p2p/conn"
)
@@ -35,10 +36,8 @@ type Peer interface {
Status() tmconn.ConnectionStatus
SocketAddr() *NetAddress // actual address of the socket
/*
NewSend(Envelope) bool
NewTrySend(Envelope) bool
*/
NewSend(Envelope) bool
NewTrySend(Envelope) bool
Send(byte, []byte) bool
TrySend(byte, []byte) bool
@@ -253,7 +252,7 @@ func (p *peer) Status() tmconn.ConnectionStatus {
return p.mconn.Status()
}
// Send msg bytes to the channel identified by chID byte. Returns false if the
// NewSend msg bytes to the channel identified by chID byte. Returns false if the
// send queue is full after timeout, specified by MConnection.
func (p *peer) NewSend(e Envelope) bool {
if !p.IsRunning() {
@@ -299,6 +298,31 @@ func (p *peer) Send(chID byte, msgBytes []byte) bool {
return res
}
// NewSend msg bytes to the channel identified by chID byte. Returns false if the
// send queue is full after timeout, specified by MConnection.
func (p *peer) NewTrySend(e Envelope) bool {
if !p.IsRunning() {
// see Switch#Broadcast, where we fetch the list of peers and loop over
// them - while we're looping, one peer may be removed and stopped.
return false
} else if !p.hasChannel(e.ChannelID) {
return false
}
msgBytes, err := proto.Marshal(e.Message)
if err != nil {
panic(err)
}
res := p.mconn.TrySend(e.ChannelID, msgBytes)
if res {
labels := []string{
"peer_id", string(p.ID()),
"chID", fmt.Sprintf("%#x", e.ChannelID),
}
p.metrics.PeerSendBytesTotal.With(labels...).Add(float64(len(msgBytes)))
}
return res
}
// TrySend msg bytes to the channel identified by chID byte. Immediately returns
// false if the send queue is full.
func (p *peer) TrySend(chID byte, msgBytes []byte) bool {
p2p/switch.go
+35
View File
@@ -6,6 +6,7 @@ import (
"sync"
"time"
"github.com/cosmos/gogoproto/proto"
"github.com/tendermint/tendermint/config"
"github.com/tendermint/tendermint/libs/cmap"
"github.com/tendermint/tendermint/libs/log"
@@ -285,6 +286,40 @@ func (sw *Switch) Broadcast(chID byte, msgBytes []byte) chan bool {
return successChan
}
// NewBroadcast runs a go routine for each attempted send, which will block trying
// to send for defaultSendTimeoutSeconds. Returns a channel which receives
// success values for each attempted send (false if times out). Channel will be
// closed once msg bytes are sent to all peers (or time out).
//
// NOTE: Broadcast uses goroutines, so order of broadcast may not be preserved.
func (sw *Switch) NewBroadcast(e Envelope) chan bool {
msgBytes, err := proto.Marshal(e.Message)
if err != nil {
panic(err)
}
sw.Logger.Debug("Broadcast", "channel", e.ChannelID, "msgBytes", log.NewLazySprintf("%X", msgBytes))
peers := sw.peers.List()
var wg sync.WaitGroup
wg.Add(len(peers))
successChan := make(chan bool, len(peers))
for _, peer := range peers {
go func(p Peer) {
defer wg.Done()
success := p.Send(e.ChannelID, msgBytes)
successChan <- success
}(peer)
}
go func() {
wg.Wait()
close(successChan)
}()
return successChan
}
// NumPeers returns the count of outbound/inbound and outbound-dialing peers.
// unconditional peers are not counted here.
func (sw *Switch) NumPeers() (outbound, inbound, dialing int) {