mirror of
https://github.com/tendermint/tendermint.git
synced 2025-12-23 14:25:19 +00:00
bdedf2ec20
* p2p: add a per-message type send and receive metric (#9622) * p2p: ressurrect the p2p envelope and use to calculate message metric Add new SendEnvelope, TrySendEnvelope, BroadcastEnvelope, and ReceiveEnvelope methods in the p2p package to work with the new envelope type. Care was taken to ensure this was performed in a non-breaking manner. Co-authored-by: William Banfield <4561443+williambanfield@users.noreply.github.com> Co-authored-by: William Banfield <wbanfield@gmail.com>
128 lines
3.2 KiB
Go
128 lines
3.2 KiB
Go
package v2
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"github.com/tendermint/tendermint/p2p"
|
|
bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
|
|
"github.com/tendermint/tendermint/state"
|
|
"github.com/tendermint/tendermint/types"
|
|
)
|
|
|
|
type iIO interface {
|
|
sendBlockRequest(peerID p2p.ID, height int64) error
|
|
sendBlockToPeer(block *types.Block, peerID p2p.ID) error
|
|
sendBlockNotFound(height int64, peerID p2p.ID) error
|
|
sendStatusResponse(base, height int64, peerID p2p.ID) error
|
|
|
|
broadcastStatusRequest()
|
|
|
|
trySwitchToConsensus(state state.State, skipWAL bool) bool
|
|
}
|
|
|
|
type switchIO struct {
|
|
sw *p2p.Switch
|
|
}
|
|
|
|
func newSwitchIo(sw *p2p.Switch) *switchIO {
|
|
return &switchIO{
|
|
sw: sw,
|
|
}
|
|
}
|
|
|
|
const (
|
|
// BlockchainChannel is a channel for blocks and status updates (`BlockStore` height)
|
|
BlockchainChannel = byte(0x40)
|
|
)
|
|
|
|
type consensusReactor interface {
|
|
// for when we switch from blockchain reactor and fast sync to
|
|
// the consensus machine
|
|
SwitchToConsensus(state state.State, skipWAL bool)
|
|
}
|
|
|
|
func (sio *switchIO) sendBlockRequest(peerID p2p.ID, height int64) error {
|
|
peer := sio.sw.Peers().Get(peerID)
|
|
if peer == nil {
|
|
return fmt.Errorf("peer not found")
|
|
}
|
|
if queued := p2p.TrySendEnvelopeShim(peer, p2p.Envelope{ //nolint: staticcheck
|
|
ChannelID: BlockchainChannel,
|
|
Message: &bcproto.BlockRequest{Height: height},
|
|
}, sio.sw.Logger); !queued {
|
|
return fmt.Errorf("send queue full")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (sio *switchIO) sendStatusResponse(base int64, height int64, peerID p2p.ID) error {
|
|
peer := sio.sw.Peers().Get(peerID)
|
|
if peer == nil {
|
|
return fmt.Errorf("peer not found")
|
|
}
|
|
|
|
if queued := p2p.TrySendEnvelopeShim(peer, p2p.Envelope{ //nolint: staticcheck
|
|
ChannelID: BlockchainChannel,
|
|
Message: &bcproto.StatusRequest{},
|
|
}, sio.sw.Logger); !queued {
|
|
return fmt.Errorf("peer queue full")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (sio *switchIO) sendBlockToPeer(block *types.Block, peerID p2p.ID) error {
|
|
peer := sio.sw.Peers().Get(peerID)
|
|
if peer == nil {
|
|
return fmt.Errorf("peer not found")
|
|
}
|
|
if block == nil {
|
|
panic("trying to send nil block")
|
|
}
|
|
|
|
bpb, err := block.ToProto()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if queued := p2p.TrySendEnvelopeShim(peer, p2p.Envelope{ //nolint: staticcheck
|
|
ChannelID: BlockchainChannel,
|
|
Message: &bcproto.BlockResponse{Block: bpb},
|
|
}, sio.sw.Logger); !queued {
|
|
return fmt.Errorf("peer queue full")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (sio *switchIO) sendBlockNotFound(height int64, peerID p2p.ID) error {
|
|
peer := sio.sw.Peers().Get(peerID)
|
|
if peer == nil {
|
|
return fmt.Errorf("peer not found")
|
|
}
|
|
if queued := p2p.TrySendEnvelopeShim(peer, p2p.Envelope{ //nolint: staticcheck
|
|
ChannelID: BlockchainChannel,
|
|
Message: &bcproto.NoBlockResponse{Height: height},
|
|
}, sio.sw.Logger); !queued {
|
|
return fmt.Errorf("peer queue full")
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (sio *switchIO) trySwitchToConsensus(state state.State, skipWAL bool) bool {
|
|
conR, ok := sio.sw.Reactor("CONSENSUS").(consensusReactor)
|
|
if ok {
|
|
conR.SwitchToConsensus(state, skipWAL)
|
|
}
|
|
return ok
|
|
}
|
|
|
|
func (sio *switchIO) broadcastStatusRequest() {
|
|
// XXX: maybe we should use an io specific peer list here
|
|
sio.sw.BroadcastEnvelope(p2p.Envelope{
|
|
ChannelID: BlockchainChannel,
|
|
Message: &bcproto.StatusRequest{},
|
|
})
|
|
}
|