Replace fnv32 with xxh3 for shard hashing

* refactors duplicated hashing function out into `helper.HashString`
* adds benchmark test coverage for `helper.HashString`
* replaces `fnv32` with `xxh3` from `github.com/zeebo/xxh3`
* replaces modulo division in `cache.GetShard` and
  `persister.Whisper.store` with a bitmask operation
* updates vendored modules
* updates carbonserver tests to force explicit ordering to avoid issues
  from arbitrary insert order changes with hash/shard assignment

Author: drawks

cache/cache.go

@@ -27,7 +27,7 @@ const (
 	Noop
 )
 
-const shardCount = 1024
+const shardCount = 1 << 10 // 1024 - an arbitrary sized power of 2
 
 type cacheSettings struct {
 	maxSize     int64
@@ -187,22 +187,9 @@ func (c *Cache) Stat(send helper.StatCallback) {
 	helper.SendAndSubstractUint32("droppedRealtimeIndex", &c.stat.droppedRealtimeIndex, send)
 }
 
-// hash function
-// @TODO: try crc32 or something else?
-func fnv32(key string) uint32 {
-	hash := uint32(2166136261)
-	const prime32 = uint32(16777619)
-	for i := 0; i < len(key); i++ {
-		hash *= prime32
-		hash ^= uint32(key[i])
-	}
-	return hash
-}
-
 // GetShard returns shard under given key
 func (c *Cache) GetShard(key string) *Shard {
-	// @TODO: remove type casts?
-	return c.data[uint(fnv32(key))%uint(shardCount)]
+	return c.data[helper.HashString(key)&(shardCount-1)]
 }
 
 func (c *Cache) Get(key string) []points.Point {

carbonserver/cache_index_test.go

@@ -137,6 +137,9 @@ func (f *testInfo) checkExpandGlobs(t *testing.T, query string, shdExist bool) {
 func (f *testInfo) commonCacheIdxTestHelper(t *testing.T) {
 	defer os.RemoveAll(f.whisperDir)
 
+	// force "sorted" strategy to ensure determinism in tests
+	f.testCache.SetWriteStrategy("sorted")
+
 	for _, filePath := range addFiles {
 		if err := addFileToSys(filePath, f.whisperDir); err != nil {
 			t.Fatal(err)
@@ -148,9 +151,11 @@ func (f *testInfo) commonCacheIdxTestHelper(t *testing.T) {
 	f.forceChan <- struct{}{}
 	time.Sleep(2 * time.Second)
 
-	// add metrics to cache
+	// add metrics to cache with distinct timestamps
+	timestamps := []int64{20, 30, 40, 50, 10}
+
 	for i, metricName := range addMetrics {
-		f.testCache.Add(points.OnePoint(metricName, float64(i), 10))
+		f.testCache.Add(points.OnePoint(metricName, float64(i), timestamps[i]))
 	}
 	// check expandblobs for new metrics
 	f.checkExpandGlobs(t, addMetrics[2], false)
@@ -161,7 +166,8 @@ func (f *testInfo) commonCacheIdxTestHelper(t *testing.T) {
 	p1, _ := f.testCache.PopNotConfirmed(m1)
 	f.testCache.Confirm(p1)
 
-	if !p1.Eq(points.OnePoint(addMetrics[4], 4, 10)) {
+	// assert [4] comes first, it has the smallest timestamp
+	if !p1.Eq(points.OnePoint(addMetrics[4], 4, timestamps[4])) {
 		fmt.Printf("error - recived wrong point - %v\n", p1)
 		t.FailNow()
 	}
@@ -183,8 +189,8 @@ func (f *testInfo) commonCacheIdxTestHelper(t *testing.T) {
 	p2, _ := f.testCache.PopNotConfirmed(m2)
 	f.testCache.Confirm(p2)
 
-	// queue within cache is sorted by length of metric name
-	if !p2.Eq(points.OnePoint(addMetrics[0], 0, 10)) {
+	// assert [0] comes second, it has the second smallest timestamp
+	if !p2.Eq(points.OnePoint(addMetrics[0], 0, timestamps[0])) {
 		fmt.Printf("error - recived wrong point - %v\n", p2)
 		t.FailNow()
 	}

go.mod

@@ -38,6 +38,7 @@ require (
 require (
 	github.com/cespare/xxhash/v2 v2.3.0
 	github.com/greatroar/blobloom v0.8.1
+	github.com/zeebo/xxh3 v1.0.2
 	golang.org/x/net v0.43.0
 	google.golang.org/protobuf v1.36.11
 )
@@ -67,6 +68,7 @@ require (
 	github.com/jcmturner/gofork v1.7.6 // indirect
 	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
 	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
+	github.com/klauspost/cpuid/v2 v2.0.9 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/onsi/ginkgo v1.14.0 // indirect
 	github.com/onsi/gomega v1.10.1 // indirect

go.sum

@@ -165,6 +165,8 @@ github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+o
 github.com/klauspost/compress v1.12.2/go.mod h1:8dP1Hq4DHOhN9w426knH3Rhby4rFm6D8eO+e+Dq5Gzg=
 github.com/klauspost/compress v1.18.2 h1:iiPHWW0YrcFgpBYhsA6D1+fqHssJscY/Tm/y2Uqnapk=
 github.com/klauspost/compress v1.18.2/go.mod h1:R0h/fSBs8DE4ENlcrlib3PsXS61voFxhIs2DeRhCvJ4=
+github.com/klauspost/cpuid/v2 v2.0.9 h1:lgaqFMSdTdQYdZ04uHyN2d/eKdOMyi2YLSvlQIBFYa4=
+github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
@@ -243,6 +245,10 @@ github.com/xdg/stringprep v1.0.3/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+github.com/zeebo/assert v1.3.0 h1:g7C04CbJuIDKNPFHmsk4hwZDO5O+kntRxzaUoNXj+IQ=
+github.com/zeebo/assert v1.3.0/go.mod h1:Pq9JiuJQpG8JLJdtkwrJESF0Foym2/D9XMU5ciN/wJ0=
+github.com/zeebo/xxh3 v1.0.2 h1:xZmwmqxHZA8AI603jOQ0tMqmBr9lPeFwGg6d+xy9DC0=
+github.com/zeebo/xxh3 v1.0.2/go.mod h1:5NWz9Sef7zIDm2JHfFlcQvNekmcEl9ekUZQQKCYaDcA=
 go.einride.tech/aip v0.73.0 h1:bPo4oqBo2ZQeBKo4ZzLb1kxYXTY1ysJhpvQyfuGzvps=
 go.einride.tech/aip v0.73.0/go.mod h1:Mj7rFbmXEgw0dq1dqJ7JGMvYCZZVxmGOR3S4ZcV5LvQ=
 go.opencensus.io v0.24.0 h1:y73uSU6J157QMP2kn2r30vwW1A2W2WFwSCGnAVxeaD0=

helper/hash.go

@@ -0,0 +1,11 @@
+package helper
+
+import (
+	"github.com/zeebo/xxh3"
+)
+
+// HashString is a wrapper for a fast 64-bit non-cryptographic hash function,
+// typically used for sharding, distribution, and other non-security-critical purposes.
+func HashString(s string) uint64 {
+	return xxh3.HashString(s)
+}

helper/hash_test.go

@@ -0,0 +1,23 @@
+package helper
+
+import (
+	"fmt"
+	"runtime"
+	"testing"
+)
+
+func BenchmarkHashString(b *testing.B) {
+	for _, n := range []int{16, 17, 32, 33, 64, 65, 96, 97, 128, 129, 240, 241} {
+		b.Run(fmt.Sprintf("%d", n), func(b *testing.B) {
+			var acc uint64
+			d := string(make([]byte, n))
+
+			b.SetBytes(int64(n))
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				acc = HashString(d)
+			}
+			runtime.KeepAlive(acc)
+		})
+	}
+}

persister/whisper.go

@@ -25,7 +25,7 @@ import (
 	"github.com/lomik/zapwriter"
 )
 
-const storeMutexCount = 32768
+const storeMutexCount = 1 << 15 // 32768
 const maxPathLength = 4095
 const maxFilenameLength = 255
 
@@ -205,15 +205,6 @@ func (p *Whisper) SetTaggedFn(fn func(string, bool)) {
 	p.taggedFn = fn
 }
 
-func fnv32(key string) uint32 {
-	hash := uint32(2166136261)
-	const prime32 = uint32(16777619)
-	for i := 0; i < len(key); i++ {
-		hash *= prime32
-		hash ^= uint32(key[i])
-	}
-	return hash
-}
 func (p *Whisper) registerOutOfOrderWriteLags(points []*whisper.TimeSeriesPoint) {
 	if !p.prometheus.enabled {
 		return
@@ -262,7 +253,7 @@ func (p *Whisper) store(metric string) {
 	// avoid concurrent store same metric
 	// @TODO: may be flock?
 	// start := time.Now()
-	mutexIndex := fnv32(metric) % storeMutexCount
+	mutexIndex := helper.HashString(metric) & (storeMutexCount - 1)
 	p.storeMutex[mutexIndex].Lock()
 	// atomic.AddUint64(&p.blockAvoidConcurrentNs, uint64(time.Since(start).Nanoseconds()))
 	defer p.storeMutex[mutexIndex].Unlock()