Reduce memory allocations in handlePacket

This converts the byte buffer to a string by casting and parses the
string by looping over it instead of calling Split.

This usage of unsafe is taken from the strings.Builder package here:
cb5c82bc3d/src/strings/builder.go (L47)

Signed-off-by: Clayton O'Neill <claytono@github.com>
This commit is contained in:
Clayton O'Neill 2019-05-16 16:32:59 -04:00
parent a294491e0b
commit 4cf6f74dc3
No known key found for this signature in database
GPG key ID: 5017D45C788B5274
2 changed files with 70 additions and 5 deletions

View file

@ -315,3 +315,54 @@ func TestHandlePacket(t *testing.T) {
}
}
}
func BenchmarkUDPListener(b *testing.B) {
scenarios := []struct {
name string
metrics [][]byte
}{
{
name: "simple counter",
metrics: [][]byte{[]byte("counter:2|c")},
}, {
name: "simple gauge",
metrics: [][]byte{[]byte("gauge:10|g")},
}, {
name: "simple timing",
metrics: [][]byte{[]byte("timing:200|ms")},
}, {
name: "simple histogram",
metrics: [][]byte{[]byte("histogram:200|h")},
}, {
name: "simple distribution",
metrics: [][]byte{[]byte("distribution:200|d")},
}, {
name: "simple_tags",
metrics: [][]byte{[]byte("simple_tags:100|c|#tag1:bar,tag2:baz")},
}, {
name: "datadog tag extension with complex tags",
metrics: [][]byte{[]byte("foo:100|c|#09digits:0,tag.with.dots:1")},
}, {
name: "datadog many tags",
metrics: [][]byte{[]byte("cpu_throttle_time_ms:1.1|ms|#action:test,application:testapp,application_component:testcomp,application_role:test_role,category:category,controller:controller,deployed_to:production,kube_deployment:deploy,kube_namespace:kube-production,method:get,version:rails5_2,status:200,status_range:2xx)}")},
}, {
name: "datadog tag extension with sampling",
metrics: [][]byte{[]byte("foo:100|c|@0.1|#tag1:bar,#tag2:baz")},
}, {
name: "combined multiline metrics",
metrics: [][]byte{[]byte("foo:200|ms:300|ms:5|c|@0.1:6|g\nbar:1|c:5|ms")},
},
}
for _, s := range scenarios {
l := &StatsDUDPListener{}
b.Run(s.name, func(b *testing.B) {
events := make(chan Events, len(s.metrics)*b.N)
defer close(events)
for i := 1; i <= b.N; i++ {
for _, m := range s.metrics {
l.handlePacket(m, events)
}
}
})
}
}

View file

@ -26,6 +26,7 @@ import (
"strings"
"time"
"unicode/utf8"
"unsafe"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/common/log"
@ -714,6 +715,7 @@ func parseDogStatsDTagsToLabels(component string) map[string]string {
}
func lineToEvents(line string) Events {
linesReceived.Inc()
events := Events{}
if line == "" {
return events
@ -836,12 +838,26 @@ func (l *StatsDUDPListener) Listen(e chan<- Events) {
func (l *StatsDUDPListener) handlePacket(packet []byte, e chan<- Events) {
udpPackets.Inc()
lines := strings.Split(string(packet), "\n")
events := Events{}
for _, line := range lines {
linesReceived.Inc()
// Convert the []byte array into a string w/o copying or allocating new
// memory, then walk the string looking for newlines to avoid memory
// allocation from Split.
p := *(*string)(unsafe.Pointer(&packet))
offset := 0
for i, c := range p {
if c == '\n' {
line := p[offset:i]
events = append(events, lineToEvents(line)...)
offset = i + 1
}
}
if offset < len(p) {
line := p[offset:]
events = append(events, lineToEvents(line)...)
}
e <- events
}
@ -884,7 +900,6 @@ func (l *StatsDTCPListener) handleConn(c *net.TCPConn, e chan<- Events) {
log.Debugf("Read %s failed: line too long", c.RemoteAddr())
break
}
linesReceived.Inc()
e <- lineToEvents(string(line))
}
}
@ -914,7 +929,6 @@ func (l *StatsDUnixgramListener) handlePacket(packet []byte, e chan<- Events) {
lines := strings.Split(string(packet), "\n")
events := Events{}
for _, line := range lines {
linesReceived.Inc()
events = append(events, lineToEvents(line)...)
}
e <- events