Skip to content

Commit

Permalink
p2p: add a per-message type send and receive metric (backport tenderm…
Browse files Browse the repository at this point in the history 
…int#9622) (tendermint#9641)

* p2p: add a per-message type send and receive metric (tendermint#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 <[email protected]>
Co-authored-by: William Banfield <[email protected]>
  • Loading branch information
@mergify @williambanfield
3 people authored Nov 1, 2022
1 parent 6e7fa2a commit bdedf2e
Show file tree
Hide file tree
Showing 54 changed files with 2,177 additions and 1,538 deletions.
3 changes: 3 additions & 0 deletions .github/workflows/lint.yml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,9 @@ jobs:
go.sum
- uses: golangci/golangci-lint-action@v3
with:
# Required: the version of golangci-lint is required and
# must be specified without patch version: we always use the
# latest patch version.
version: v1.50.1
args: --timeout 10m
github-token: ${{ secrets.github_token }}
Expand Down
2 changes: 2 additions & 0 deletions CHANGELOG_PENDING.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,5 +17,7 @@
### FEATURES

### IMPROVEMENTS
- [p2p] \#9641 Add new Envelope type and associated methods for sending and receiving Envelopes instead of raw bytes.
This also adds new metrics, `tendermint_p2p_message_send_bytes_total` and `tendermint_p2p_message_receive_bytes_total`, that expose how many bytes of each message type have been sent.

### BUG FIXES
2 changes: 1 addition & 1 deletion Makefile
View file
Original file line number Diff line number Diff line change
Expand Up @@ -200,7 +200,7 @@ format:

lint:
@echo "--> Running linter"
@golangci-lint run
@go run github.com/golangci/golangci-lint/cmd/golangci-lint run
.PHONY: lint

DESTINATION = ./index.html.md
Expand Down
81 changes: 29 additions & 52 deletions blockchain/msgs.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ import (

"github.com/gogo/protobuf/proto"

"github.com/tendermint/tendermint/p2p"
bcproto "github.com/tendermint/tendermint/proto/tendermint/blockchain"
"github.com/tendermint/tendermint/types"
)
Expand All @@ -19,58 +20,6 @@ const (
BlockResponseMessageFieldKeySize
)

// EncodeMsg encodes a Protobuf message
func EncodeMsg(pb proto.Message) ([]byte, error) {
msg := bcproto.Message{}

switch pb := pb.(type) {
case *bcproto.BlockRequest:
msg.Sum = &bcproto.Message_BlockRequest{BlockRequest: pb}
case *bcproto.BlockResponse:
msg.Sum = &bcproto.Message_BlockResponse{BlockResponse: pb}
case *bcproto.NoBlockResponse:
msg.Sum = &bcproto.Message_NoBlockResponse{NoBlockResponse: pb}
case *bcproto.StatusRequest:
msg.Sum = &bcproto.Message_StatusRequest{StatusRequest: pb}
case *bcproto.StatusResponse:
msg.Sum = &bcproto.Message_StatusResponse{StatusResponse: pb}
default:
return nil, fmt.Errorf("unknown message type %T", pb)
}

bz, err := proto.Marshal(&msg)
if err != nil {
return nil, fmt.Errorf("unable to marshal %T: %w", pb, err)
}

return bz, nil
}

// DecodeMsg decodes a Protobuf message.
func DecodeMsg(bz []byte) (proto.Message, error) {
pb := &bcproto.Message{}

err := proto.Unmarshal(bz, pb)
if err != nil {
return nil, err
}

switch msg := pb.Sum.(type) {
case *bcproto.Message_BlockRequest:
return msg.BlockRequest, nil
case *bcproto.Message_BlockResponse:
return msg.BlockResponse, nil
case *bcproto.Message_NoBlockResponse:
return msg.NoBlockResponse, nil
case *bcproto.Message_StatusRequest:
return msg.StatusRequest, nil
case *bcproto.Message_StatusResponse:
return msg.StatusResponse, nil
default:
return nil, fmt.Errorf("unknown message type %T", msg)
}
}

// ValidateMsg validates a message.
func ValidateMsg(pb proto.Message) error {
if pb == nil {
Expand Down Expand Up @@ -108,3 +57,31 @@ func ValidateMsg(pb proto.Message) error {
}
return nil
}

// EncodeMsg encodes a Protobuf message
//
// Deprecated: Will be removed in v0.37.
func EncodeMsg(pb proto.Message) ([]byte, error) {
if um, ok := pb.(p2p.Wrapper); ok {
pb = um.Wrap()
}
bz, err := proto.Marshal(pb)
if err != nil {
return nil, fmt.Errorf("unable to marshal %T: %w", pb, err)
}

return bz, nil
}

// DecodeMsg decodes a Protobuf message.
//
// Deprecated: Will be removed in v0.37.
func DecodeMsg(bz []byte) (proto.Message, error) {
pb := &bcproto.Message{}

err := proto.Unmarshal(bz, pb)
if err != nil {
return nil, err
}
return pb.Unwrap()
}
128 changes: 60 additions & 68 deletions blockchain/v0/reactor.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,8 @@ import (
"reflect"
"time"

"github.com/gogo/protobuf/proto"

bc "github.com/tendermint/tendermint/blockchain"
"github.com/tendermint/tendermint/libs/log"
"github.com/tendermint/tendermint/p2p"
Expand Down Expand Up @@ -144,21 +146,20 @@ func (bcR *BlockchainReactor) GetChannels() []*p2p.ChannelDescriptor {
SendQueueCapacity: 1000,
RecvBufferCapacity: 50 * 4096,
RecvMessageCapacity: bc.MaxMsgSize,
MessageType: &bcproto.Message{},
},
}
}

// AddPeer implements Reactor by sending our state to peer.
func (bcR *BlockchainReactor) AddPeer(peer p2p.Peer) {
msgBytes, err := bc.EncodeMsg(&bcproto.StatusResponse{
Base: bcR.store.Base(),
Height: bcR.store.Height()})
if err != nil {
bcR.Logger.Error("could not convert msg to protobuf", "err", err)
return
}

peer.Send(BlockchainChannel, msgBytes)
p2p.SendEnvelopeShim(peer, p2p.Envelope{ //nolint: staticcheck
ChannelID: BlockchainChannel,
Message: &bcproto.StatusResponse{
Base: bcR.store.Base(),
Height: bcR.store.Height(),
},
}, bcR.Logger)
// it's OK if send fails. will try later in poolRoutine

// peer is added to the pool once we receive the first
Expand All @@ -182,75 +183,73 @@ func (bcR *BlockchainReactor) respondToPeer(msg *bcproto.BlockRequest,
bcR.Logger.Error("could not convert msg to protobuf", "err", err)
return false
}

msgBytes, err := bc.EncodeMsg(&bcproto.BlockResponse{Block: bl})
if err != nil {
bcR.Logger.Error("could not marshal msg", "err", err)
return false
}

return src.TrySend(BlockchainChannel, msgBytes)
}

bcR.Logger.Info("Peer asking for a block we don't have", "src", src, "height", msg.Height)

msgBytes, err := bc.EncodeMsg(&bcproto.NoBlockResponse{Height: msg.Height})
if err != nil {
bcR.Logger.Error("could not convert msg to protobuf", "err", err)
return false
return p2p.TrySendEnvelopeShim(src, p2p.Envelope{ //nolint: staticcheck
ChannelID: BlockchainChannel,
Message: &bcproto.BlockResponse{Block: bl},
}, bcR.Logger)
}

return src.TrySend(BlockchainChannel, msgBytes)
return p2p.TrySendEnvelopeShim(src, p2p.Envelope{ //nolint: staticcheck
ChannelID: BlockchainChannel,
Message: &bcproto.NoBlockResponse{Height: msg.Height},
}, bcR.Logger)
}

// Receive implements Reactor by handling 4 types of messages (look below).
func (bcR *BlockchainReactor) Receive(chID byte, src p2p.Peer, msgBytes []byte) {
msg, err := bc.DecodeMsg(msgBytes)
if err != nil {
bcR.Logger.Error("Error decoding message", "src", src, "chId", chID, "err", err)
bcR.Switch.StopPeerForError(src, err)
func (bcR *BlockchainReactor) ReceiveEnvelope(e p2p.Envelope) {
if err := bc.ValidateMsg(e.Message); err != nil {
bcR.Logger.Error("Peer sent us invalid msg", "peer", e.Src, "msg", e.Message, "err", err)
bcR.Switch.StopPeerForError(e.Src, err)
return
}

if err = bc.ValidateMsg(msg); err != nil {
bcR.Logger.Error("Peer sent us invalid msg", "peer", src, "msg", msg, "err", err)
bcR.Switch.StopPeerForError(src, err)
return
}
bcR.Logger.Debug("Receive", "e.Src", e.Src, "chID", e.ChannelID, "msg", e.Message)

bcR.Logger.Debug("Receive", "src", src, "chID", chID, "msg", msg)

switch msg := msg.(type) {
switch msg := e.Message.(type) {
case *bcproto.BlockRequest:
bcR.respondToPeer(msg, src)
bcR.respondToPeer(msg, e.Src)
case *bcproto.BlockResponse:
bi, err := types.BlockFromProto(msg.Block)
if err != nil {
bcR.Logger.Error("Block content is invalid", "err", err)
return
}
bcR.pool.AddBlock(src.ID(), bi, len(msgBytes))
bcR.pool.AddBlock(e.Src.ID(), bi, msg.Block.Size())
case *bcproto.StatusRequest:
// Send peer our state.
msgBytes, err := bc.EncodeMsg(&bcproto.StatusResponse{
Height: bcR.store.Height(),
Base: bcR.store.Base(),
})
if err != nil {
bcR.Logger.Error("could not convert msg to protobut", "err", err)
return
}
src.TrySend(BlockchainChannel, msgBytes)
p2p.TrySendEnvelopeShim(e.Src, p2p.Envelope{ //nolint: staticcheck
ChannelID: BlockchainChannel,
Message: &bcproto.StatusResponse{
Height: bcR.store.Height(),
Base: bcR.store.Base(),
},
}, bcR.Logger)
case *bcproto.StatusResponse:
// Got a peer status. Unverified.
bcR.pool.SetPeerRange(src.ID(), msg.Base, msg.Height)
bcR.pool.SetPeerRange(e.Src.ID(), msg.Base, msg.Height)
case *bcproto.NoBlockResponse:
bcR.Logger.Debug("Peer does not have requested block", "peer", src, "height", msg.Height)
bcR.Logger.Debug("Peer does not have requested block", "peer", e.Src, "height", msg.Height)
default:
bcR.Logger.Error(fmt.Sprintf("Unknown message type %v", reflect.TypeOf(msg)))
}
}

func (bcR *BlockchainReactor) Receive(chID byte, peer p2p.Peer, msgBytes []byte) {
var msg *bcproto.Message
err := proto.Unmarshal(msgBytes, msg)
if err != nil {
panic(err)
}
uw, err := msg.Unwrap()
if err != nil {
panic(err)
}
bcR.ReceiveEnvelope(p2p.Envelope{
ChannelID: chID,
Src: peer,
Message: uw,
})
}

// Handle messages from the poolReactor telling the reactor what to do.
// NOTE: Don't sleep in the FOR_LOOP or otherwise slow it down!
func (bcR *BlockchainReactor) poolRoutine(stateSynced bool) {
Expand Down Expand Up @@ -286,13 +285,10 @@ func (bcR *BlockchainReactor) poolRoutine(stateSynced bool) {
if peer == nil {
continue
}
msgBytes, err := bc.EncodeMsg(&bcproto.BlockRequest{Height: request.Height})
if err != nil {
bcR.Logger.Error("could not convert msg to proto", "err", err)
continue
}

queued := peer.TrySend(BlockchainChannel, msgBytes)
queued := p2p.TrySendEnvelopeShim(peer, p2p.Envelope{ //nolint: staticcheck
ChannelID: BlockchainChannel,
Message: &bcproto.BlockRequest{Height: request.Height},
}, bcR.Logger)
if !queued {
bcR.Logger.Debug("Send queue is full, drop block request", "peer", peer.ID(), "height", request.Height)
}
Expand Down Expand Up @@ -425,13 +421,9 @@ FOR_LOOP:

// BroadcastStatusRequest broadcasts `BlockStore` base and height.
func (bcR *BlockchainReactor) BroadcastStatusRequest() error {
bm, err := bc.EncodeMsg(&bcproto.StatusRequest{})
if err != nil {
bcR.Logger.Error("could not convert msg to proto", "err", err)
return fmt.Errorf("could not convert msg to proto: %w", err)
}

bcR.Switch.Broadcast(BlockchainChannel, bm)

bcR.Switch.BroadcastEnvelope(p2p.Envelope{
ChannelID: BlockchainChannel,
Message: &bcproto.StatusRequest{},
})
return nil
}
Loading

0 comments on commit bdedf2e

Please sign in to comment.