p2p: cache per-message ingress gauges to avoid allocs in Peer.handle

Author: AskAlexSharov

p2p/peer.go

@@ -364,9 +364,7 @@ func (p *Peer) handle(msg Msg) error {
 		}
 
 		if p.metricsEnabled {
-			localCode := msg.Code - proto.offset
-			proto.meters[localCode].SetUint32(msg.meterSize)
-			proto.meterPackets[localCode].Set(1)
+			proto.meterIngress(msg.Code-proto.offset, msg.meterSize)
 		}
 		select {
 		case proto.in <- msg:
@@ -468,11 +466,26 @@ type protoRW struct {
 	w      MsgWriter
 	logger log.Logger
 
-	// cached gauge pointers indexed by local message code, avoids lock+map lookup on every message
 	meters       []metrics.Gauge
 	meterPackets []metrics.Gauge
 }
 
+func (rw *protoRW) meterIngress(code uint64, size uint32) {
+	if rw.meters == nil {
+		rw.meters = make([]metrics.Gauge, rw.Length)
+		rw.meterPackets = make([]metrics.Gauge, rw.Length)
+	}
+	g := rw.meters[code]
+	if g == nil {
+		name := fmt.Sprintf("%s_%s_%d_%#02x", ingressMeterName, rw.Name, rw.Version, code)
+		g = metrics.GetOrCreateGauge(name)
+		rw.meters[code] = g
+		rw.meterPackets[code] = metrics.GetOrCreateGauge(name + "_packets")
+	}
+	g.SetUint32(size)
+	rw.meterPackets[code].Set(1)
+}
+
 var traceMsg = false
 
 func (rw *protoRW) WriteMsg(msg Msg) (err error) {

p2p/peer_test.go

@@ -32,6 +32,7 @@ import (
 	"time"
 
 	"github.com/erigontech/erigon/common/log/v3"
+	"github.com/erigontech/erigon/diagnostics/metrics"
 	"github.com/erigontech/erigon/p2p/enode"
 	"github.com/erigontech/erigon/p2p/enr"
 )
@@ -147,6 +148,30 @@ func TestPeerProtoReadMsg(t *testing.T) {
 	}
 }
 
+func TestProtoRWMeterIngress(t *testing.T) {
+	rw := &protoRW{Protocol: Protocol{Name: "metertest", Version: 99, Length: 4}}
+
+	rw.meterIngress(2, 1234)
+
+	if got := metrics.GetOrCreateGauge("p2p_ingress_metertest_99_0x02").GetValue(); got != 1234 {
+		t.Fatalf("ingress bytes gauge = %v, want 1234", got)
+	}
+	if got := metrics.GetOrCreateGauge("p2p_ingress_metertest_99_0x02_packets").GetValue(); got != 1 {
+		t.Fatalf("ingress packets gauge = %v, want 1", got)
+	}
+}
+
+func TestProtoRWMeterIngressNoAllocs(t *testing.T) {
+	rw := &protoRW{Protocol: Protocol{Name: "metertestnoalloc", Version: 1, Length: 8}}
+	rw.meterIngress(3, 100) // warm the per-code gauge cache
+
+	if allocs := testing.AllocsPerRun(1000, func() {
+		rw.meterIngress(3, 100)
+	}); allocs != 0 {
+		t.Errorf("meterIngress allocates %v objects per call after warmup, want 0", allocs)
+	}
+}
+
 func TestPeerProtoEncodeMsg(t *testing.T) {
 	proto := Protocol{
 		Name:   "a",