fix receive

p2p/metrics.go

@@ -25,6 +25,10 @@ type Metrics struct {
 	PeerPendingSendBytes metrics.Gauge
 	// Number of transactions submitted by each peer.
 	NumTxs metrics.Gauge
+	// Histogram of message receive duration.
+	MessageReceiveTime metrics.Histogram
+	// Histogram of message send duration.
+	MessageSendTime metrics.Histogram
 }
 
 // PrometheusMetrics returns Metrics build using Prometheus client library.
@@ -66,6 +70,22 @@ func PrometheusMetrics(namespace string, labelsAndValues ...string) *Metrics {
 			Name:      "num_txs",
 			Help:      "Number of transactions submitted by each peer.",
 		}, append(labels, "peer_id")).With(labelsAndValues...),
+		MessageReceiveTime: prometheus.NewHistogramFrom(stdprometheus.HistogramOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "message_receive_time",
+			Help:      "Histogram of message receive duration.",
+
+			Buckets: stdprometheus.ExponentialBucketsRange(0.05, 50, 8),
+		}, labels).With(labelsAndValues...),
+		MessageSendTime: prometheus.NewHistogramFrom(stdprometheus.HistogramOpts{
+			Namespace: namespace,
+			Subsystem: MetricsSubsystem,
+			Name:      "message_send_time",
+			Help:      "Histogram of message send duration.",
+
+			Buckets: stdprometheus.ExponentialBucketsRange(0.05, 50, 8),
+		}, labels).With(labelsAndValues...),
 	}
 }
 
@@ -77,5 +97,7 @@ func NopMetrics() *Metrics {
 		PeerSendBytesTotal:    discard.NewCounter(),
 		PeerPendingSendBytes:  discard.NewGauge(),
 		NumTxs:                discard.NewGauge(),
+		MessageReceiveTime:    discard.NewHistogram(),
+		MessageSendTime:       discard.NewHistogram(),
 	}
 }

p2p/peer.go

@@ -257,6 +257,10 @@ func (p *peer) Send(chID byte, msgBytes []byte) bool {
 	} else if !p.hasChannel(chID) {
 		return false
 	}
+	before := time.Now()
+	defer func() {
+		p.metrics.MessageSendTime.Observe(time.Since(before).Seconds())
+	}()
 	res := p.mconn.Send(chID, msgBytes)
 	if res {
 		labels := []string{
@@ -276,6 +280,10 @@ func (p *peer) TrySend(chID byte, msgBytes []byte) bool {
 	} else if !p.hasChannel(chID) {
 		return false
 	}
+	before := time.Now()
+	defer func() {
+		p.metrics.MessageSendTime.Observe(time.Since(before).Seconds())
+	}()
 	res := p.mconn.TrySend(chID, msgBytes)
 	if res {
 		labels := []string{
@@ -390,6 +398,10 @@ func createMConnection(
 ) *tmconn.MConnection {
 
 	onReceive := func(chID byte, msgBytes []byte) {
+		before := time.Now()
+		defer func() {
+			p.metrics.MessageReceiveTime.Observe(time.Since(before).Seconds())
+		}()
 		reactor := reactorsByCh[chID]
 		if reactor == nil {
 			// Note that its ok to panic here as it's caught in the conn._recover,