Uses forked pkg/util/interner for dogstatsd workload

Author: scottopell

comp/dogstatsd/server/intern.go

@@ -6,93 +6,92 @@
 package server
 
 import (
+	"runtime"
+	"unsafe"
+
 	"github.com/DataDog/datadog-agent/pkg/config/utils"
 	"github.com/DataDog/datadog-agent/pkg/telemetry"
-	"github.com/DataDog/datadog-agent/pkg/util/log"
 )
 
 var (
 	// There are multiple instances of the interner, one per worker. Counters are normally fine,
 	// gauges require special care to make sense. We don't need to clean up when an instance is
 	// dropped, because it only happens on agent shutdown.
-	tlmSIResets = telemetry.NewSimpleCounter("dogstatsd", "string_interner_resets",
-		"Amount of resets of the string interner used in dogstatsd")
-	tlmSIRSize = telemetry.NewSimpleGauge("dogstatsd", "string_interner_entries",
-		"Number of entries in the string interner")
-	tlmSIRBytes = telemetry.NewSimpleGauge("dogstatsd", "string_interner_bytes",
-		"Number of bytes stored in the string interner")
 	tlmSIRHits = telemetry.NewSimpleCounter("dogstatsd", "string_interner_hits",
 		"Number of times string interner returned an existing string")
 	tlmSIRMiss = telemetry.NewSimpleCounter("dogstatsd", "string_interner_miss",
 		"Number of times string interner created a new string object")
 	tlmSIRNew = telemetry.NewSimpleCounter("dogstatsd", "string_interner_new",
 		"Number of times string interner was created")
-	tlmSIRStrBytes = telemetry.NewSimpleHistogram("dogstatsd", "string_interner_str_bytes",
-		"Number of times string with specific length were added",
-		[]float64{1, 2, 4, 8, 16, 32, 64, 128})
 )
 
-// stringInterner is a string cache providing a longer life for strings,
-// helping to avoid GC runs because they're re-used many times instead of
-// created every time.
+// A StringValue pointer is the handle to the underlying string value.
+// See Get how Value pointers may be used.
+type StringValue struct {
+	_           [0]func() // prevent people from accidentally using value type as comparable
+	cmpVal      string
+	resurrected bool
+}
+
+// Get the underlying string value
+func (v *StringValue) Get() string {
+	return v.cmpVal
+}
+
+// stringInterner interns strings while allowing them to be cleaned up by the GC.
+// It can handle both string and []byte types without allocation.
 type stringInterner struct {
-	strings    map[string]string
-	maxSize    int
-	curBytes   int
 	tlmEnabled bool
+	valMap     map[string]uintptr
 }
 
-func newStringInterner(maxSize int) *stringInterner {
+// newStringInterner creates a new StringInterner
+func newStringInterner() *stringInterner {
 	i := &stringInterner{
-		strings:    make(map[string]string),
-		maxSize:    maxSize,
+		valMap:     make(map[string]uintptr),
 		tlmEnabled: utils.IsTelemetryEnabled(),
 	}
+
 	if i.tlmEnabled {
 		tlmSIRNew.Inc()
 	}
 	return i
 }
 
-// LoadOrStore always returns the string from the cache, adding it into the
-// cache if needed.
-// If we need to store a new entry and the cache is at its maximum capacity,
-// it is reset.
-func (i *stringInterner) LoadOrStore(key []byte) string {
-	// here is the string interner trick: the map lookup using
-	// string(key) doesn't actually allocate a string, but is
-	// returning the string value -> no new heap allocation
-	// for this string.
-	// See https://github.com/golang/go/commit/f5f5a8b6209f84961687d993b93ea0d397f5d5bf
-	if s, found := i.strings[string(key)]; found {
-		if i.tlmEnabled {
+// Get returns a pointer representing the []byte k
+//
+// The returned pointer will be the same for Get(v) and Get(v2)
+// if and only if v == v2. The returned pointer will also be the same
+// for a string with same contents as the byte slice.
+//
+//go:nocheckptr
+func (s *stringInterner) LoadOrStore(k []byte) *StringValue {
+	var v *StringValue
+	// the compiler will optimize the following map lookup to not alloc a string
+	if addr, ok := s.valMap[string(k)]; ok {
+		//goland:noinspection GoVetUnsafePointer
+		v = (*StringValue)((unsafe.Pointer)(addr))
+		v.resurrected = true
+		if s.tlmEnabled {
 			tlmSIRHits.Inc()
 		}
-		return s
+		return v
 	}
-	if len(i.strings) >= i.maxSize {
-		if i.tlmEnabled {
-			tlmSIResets.Inc()
-			tlmSIRBytes.Sub(float64(i.curBytes))
-			tlmSIRSize.Sub(float64(len(i.strings)))
-			i.curBytes = 0
-		}
 
-		i.strings = make(map[string]string)
-		log.Debug("clearing the string interner cache")
-
-	}
-
-	s := string(key)
-	i.strings[s] = s
+	v = &StringValue{cmpVal: string(k)}
+	runtime.SetFinalizer(v, s.finalize)
+	s.valMap[string(k)] = uintptr(unsafe.Pointer(v))
+	tlmSIRMiss.Inc()
+	return v
+}
 
-	if i.tlmEnabled {
-		tlmSIRMiss.Inc()
-		tlmSIRSize.Inc()
-		tlmSIRBytes.Add(float64(len(s)))
-		tlmSIRStrBytes.Observe(float64(len(s)))
-		i.curBytes += len(s)
+func (s *stringInterner) finalize(v *StringValue) {
+	if v.resurrected {
+		// We lost the race. Somebody resurrected it while we
+		// were about to finalize it. Try again next round.
+		v.resurrected = false
+		runtime.SetFinalizer(v, s.finalize)
+		return
 	}
-
-	return s
+	delete(s.valMap, v.Get())
 }

comp/dogstatsd/server/intern_test.go

@@ -13,7 +13,7 @@ import (
 
 func TestInternLoadOrStoreValue(t *testing.T) {
 	assert := assert.New(t)
-	sInterner := newStringInterner(3)
+	sInterner := newStringInterner()
 
 	foo := []byte("foo")
 	bar := []byte("bar")
@@ -22,59 +22,59 @@ func TestInternLoadOrStoreValue(t *testing.T) {
 
 	// first test that the good value is returned.
 
-	v := sInterner.LoadOrStore(foo)
+	v := sInterner.LoadOrStore(foo).Get()
 	assert.Equal("foo", v)
-	v = sInterner.LoadOrStore(bar)
+	v = sInterner.LoadOrStore(bar).Get()
 	assert.Equal("bar", v)
-	v = sInterner.LoadOrStore(far)
+	v = sInterner.LoadOrStore(far).Get()
 	assert.Equal("far", v)
-	v = sInterner.LoadOrStore(boo)
+	v = sInterner.LoadOrStore(boo).Get()
 	assert.Equal("boo", v)
 }
 
 func TestInternLoadOrStorePointer(t *testing.T) {
 	assert := assert.New(t)
-	sInterner := newStringInterner(4)
+	sInterner := newStringInterner()
 
 	foo := []byte("foo")
 	bar := []byte("bar")
 	boo := []byte("boo")
 
 	// first test that the good value is returned.
 
-	v := sInterner.LoadOrStore(foo)
+	v := sInterner.LoadOrStore(foo).Get()
 	assert.Equal("foo", v)
-	v2 := sInterner.LoadOrStore(foo)
+	v2 := sInterner.LoadOrStore(foo).Get()
 	assert.Equal(&v, &v2, "must point to the same address")
-	v2 = sInterner.LoadOrStore(bar)
+	v2 = sInterner.LoadOrStore(bar).Get()
 	assert.NotEqual(&v, &v2, "must point to a different address")
-	v3 := sInterner.LoadOrStore(bar)
+	v3 := sInterner.LoadOrStore(bar).Get()
 	assert.Equal(&v2, &v3, "must point to the same address")
 
-	v4 := sInterner.LoadOrStore(boo)
+	v4 := sInterner.LoadOrStore(boo).Get()
 	assert.NotEqual(&v, &v4, "must point to a different address")
 	assert.NotEqual(&v2, &v4, "must point to a different address")
 	assert.NotEqual(&v3, &v4, "must point to a different address")
 }
 
 func TestInternLoadOrStoreReset(t *testing.T) {
 	assert := assert.New(t)
-	sInterner := newStringInterner(4)
+	sInterner := newStringInterner()
 
 	// first test that the good value is returned.
 	sInterner.LoadOrStore([]byte("foo"))
-	assert.Equal(1, len(sInterner.strings))
+	assert.Equal(1, len(sInterner.valMap))
 	sInterner.LoadOrStore([]byte("bar"))
 	sInterner.LoadOrStore([]byte("bar"))
-	assert.Equal(2, len(sInterner.strings))
+	assert.Equal(2, len(sInterner.valMap))
 	sInterner.LoadOrStore([]byte("boo"))
-	assert.Equal(3, len(sInterner.strings))
+	assert.Equal(3, len(sInterner.valMap))
 	sInterner.LoadOrStore([]byte("far"))
 	sInterner.LoadOrStore([]byte("far"))
 	sInterner.LoadOrStore([]byte("far"))
-	assert.Equal(4, len(sInterner.strings))
+	assert.Equal(4, len(sInterner.valMap))
 	sInterner.LoadOrStore([]byte("val"))
-	assert.Equal(1, len(sInterner.strings))
+	assert.Equal(5, len(sInterner.valMap))
 	sInterner.LoadOrStore([]byte("val"))
-	assert.Equal(1, len(sInterner.strings))
+	assert.Equal(5, len(sInterner.valMap))
 }

comp/dogstatsd/server/parse.go

@@ -51,11 +51,10 @@ type parser struct {
 }
 
 func newParser(cfg config.Reader, float64List *float64ListPool) *parser {
-	stringInternerCacheSize := cfg.GetInt("dogstatsd_string_interner_size")
 	readTimestamps := cfg.GetBool("dogstatsd_no_aggregation_pipeline")
 
 	return &parser{
-		interner:         newStringInterner(stringInternerCacheSize),
+		interner:         newStringInterner(),
 		readTimestamps:   readTimestamps,
 		float64List:      float64List,
 		dsdOriginEnabled: cfg.GetBool("dogstatsd_origin_detection_client"),
@@ -97,11 +96,11 @@ func (p *parser) parseTags(rawTags []byte) []string {
 		if tagPos < 0 {
 			break
 		}
-		tagsList[i] = p.interner.LoadOrStore(rawTags[:tagPos])
+		tagsList[i] = p.interner.LoadOrStore(rawTags[:tagPos]).Get()
 		rawTags = rawTags[tagPos+len(commaSeparator):]
 		i++
 	}
-	tagsList[i] = p.interner.LoadOrStore(rawTags)
+	tagsList[i] = p.interner.LoadOrStore(rawTags).Get()
 	return tagsList
 }
 
@@ -188,7 +187,7 @@ func (p *parser) parseMetricSample(message []byte) (dogstatsdMetricSample, error
 	}
 
 	return dogstatsdMetricSample{
-		name:        p.interner.LoadOrStore(name),
+		name:        p.interner.LoadOrStore(name).Get(),
 		value:       value,
 		values:      values,
 		setValue:    string(setValue),