metrics: sliding histogram

eth/protocols/eth/handler.go

@@ -218,9 +218,7 @@ func handleMessage(backend Backend, peer *Peer) error {
 		h := fmt.Sprintf("%s/%s/%d/%#02x", p2p.HandleHistName, ProtocolName, peer.Version(), msg.Code)
 		defer func(start time.Time) {
 			sampler := func() metrics.Sample {
-				return metrics.ResettingSample(
-					metrics.NewExpDecaySample(1028, 0.015),
-				)
+				return metrics.NewBoundedHistogramSample()
 			}
 			metrics.GetOrRegisterHistogramLazy(h, nil, sampler).Update(time.Since(start).Microseconds())
 		}(time.Now())

eth/protocols/snap/handler.go

@@ -145,9 +145,7 @@ func HandleMessage(backend Backend, peer *Peer) error {
 		h := fmt.Sprintf("%s/%s/%d/%#02x", p2p.HandleHistName, ProtocolName, peer.Version(), msg.Code)
 		defer func(start time.Time) {
 			sampler := func() metrics.Sample {
-				return metrics.ResettingSample(
-					metrics.NewExpDecaySample(1028, 0.015),
-				)
+				return metrics.NewBoundedHistogramSample()
 			}
 			metrics.GetOrRegisterHistogramLazy(h, nil, sampler).Update(time.Since(start).Microseconds())
 		}(start)

go.mod

@@ -25,6 +25,7 @@ require (
 	github.com/fjl/gencodec v0.0.0-20230517082657-f9840df7b83e
 	github.com/fjl/memsize v0.0.0-20190710130421-bcb5799ab5e5
 	github.com/fsnotify/fsnotify v1.6.0
+	github.com/gammazero/deque v0.2.1
 	github.com/gballet/go-libpcsclite v0.0.0-20190607065134-2772fd86a8ff
 	github.com/gballet/go-verkle v0.1.1-0.20231031103413-a67434b50f46
 	github.com/gofrs/flock v0.8.1
@@ -98,7 +99,7 @@ require (
 	github.com/consensys/bavard v0.1.13 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 // indirect
-	github.com/deepmap/oapi-codegen v1.6.0 // indirect
+	github.com/deepmap/oapi-codegen v1.8.2 // indirect
 	github.com/dlclark/regexp2 v1.7.0 // indirect
 	github.com/garslo/gogen v0.0.0-20170306192744-1d203ffc1f61 // indirect
 	github.com/go-ole/go-ole v1.3.0 // indirect

go.sum

@@ -160,8 +160,9 @@ github.com/decred/dcrd/crypto/blake256 v1.0.0 h1:/8DMNYp9SGi5f0w7uCm6d6M4OU2rGFK
 github.com/decred/dcrd/crypto/blake256 v1.0.0/go.mod h1:sQl2p6Y26YV+ZOcSTP6thNdn47hh8kt6rqSlvmrXFAc=
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1 h1:YLtO71vCjJRCBcrPMtQ9nqBsqpA1m5sE92cU+pd5Mcc=
 github.com/decred/dcrd/dcrec/secp256k1/v4 v4.0.1/go.mod h1:hyedUtir6IdtD/7lIxGeCxkaw7y45JueMRL4DIyJDKs=
-github.com/deepmap/oapi-codegen v1.6.0 h1:w/d1ntwh91XI0b/8ja7+u5SvA4IFfM0UNNLmiDR1gg0=
 github.com/deepmap/oapi-codegen v1.6.0/go.mod h1:ryDa9AgbELGeB+YEXE1dR53yAjHwFvE9iAUlWl9Al3M=
+github.com/deepmap/oapi-codegen v1.8.2 h1:SegyeYGcdi0jLLrpbCMoJxnUUn8GBXHsvr4rbzjuhfU=
+github.com/deepmap/oapi-codegen v1.8.2/go.mod h1:YLgSKSDv/bZQB7N4ws6luhozi3cEdRktEqrX88CvjIw=
 github.com/dgraph-io/badger v1.6.0/go.mod h1:zwt7syl517jmP8s94KqSxTlM6IMsdhYy6psNgSztDR4=
 github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
@@ -196,6 +197,8 @@ github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMo
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
 github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=
 github.com/fsnotify/fsnotify v1.6.0/go.mod h1:sl3t1tCWJFWoRz9R8WJCbQihKKwmorjAbSClcnxKAGw=
+github.com/gammazero/deque v0.2.1 h1:qSdsbG6pgp6nL7A0+K/B7s12mcCY/5l5SIUpMOl+dC0=
+github.com/gammazero/deque v0.2.1/go.mod h1:LFroj8x4cMYCukHJDbxFCkT+r9AndaJnFMuZDV34tuU=
 github.com/garslo/gogen v0.0.0-20170306192744-1d203ffc1f61 h1:IZqZOB2fydHte3kUgxrzK5E1fW7RQGeDwE8F/ZZnUYc=
 github.com/garslo/gogen v0.0.0-20170306192744-1d203ffc1f61/go.mod h1:Q0X6pkwTILDlzrGEckF6HKjXe48EgsY/l7K7vhY4MW8=
 github.com/gavv/httpexpect v2.0.0+incompatible/go.mod h1:x+9tiU1YnrOvnB725RkpoLv1M62hOWzwo5OXotisrKc=
@@ -204,6 +207,7 @@ github.com/gballet/go-libpcsclite v0.0.0-20190607065134-2772fd86a8ff/go.mod h1:x
 github.com/gballet/go-verkle v0.1.1-0.20231031103413-a67434b50f46 h1:BAIP2GihuqhwdILrV+7GJel5lyPV3u1+PgzrWLc0TkE=
 github.com/gballet/go-verkle v0.1.1-0.20231031103413-a67434b50f46/go.mod h1:QNpY22eby74jVhqH4WhDLDwxc/vqsern6pW+u2kbkpc=
 github.com/getkin/kin-openapi v0.53.0/go.mod h1:7Yn5whZr5kJi6t+kShccXS8ae1APpYTW6yheSwk8Yi4=
+github.com/getkin/kin-openapi v0.61.0/go.mod h1:7Yn5whZr5kJi6t+kShccXS8ae1APpYTW6yheSwk8Yi4=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
 github.com/gin-contrib/sse v0.0.0-20190301062529-5545eab6dad3/go.mod h1:VJ0WA2NBN22VlZ2dKZQPAPnyWw5XTlK1KymzLKsr59s=
 github.com/gin-gonic/gin v1.4.0/go.mod h1:OW2EZn3DO8Ln9oIKOvM++LBO+5UPHJJDH72/q/3rZdM=
@@ -221,7 +225,6 @@ github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9
 github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk=
 github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A=
 github.com/go-martini/martini v0.0.0-20170121215854-22fa46961aab/go.mod h1:/P9AEU963A2AYjv4d1V5eVL1CQbEJq6aCNHDDjibzu8=
-github.com/go-ole/go-ole v1.2.5 h1:t4MGB5xEDZvXI+0rMjjsfBsD7yAgp/s9ZDkL1JndXwY=
 github.com/go-ole/go-ole v1.2.5/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
 github.com/go-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE=
 github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78=
@@ -777,8 +780,6 @@ golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.15.0 h1:h48lPFYpsTvQJZF4EKyI4aLHaev3CxivZmv7yZig9pc=
-golang.org/x/sys v0.15.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/sys v0.16.0 h1:xWw16ngr6ZMtmxDyKyIgsE93KNKz5HKmMa3b8ALHidU=
 golang.org/x/sys v0.16.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=

metrics/chunked_associative_array.go

@@ -0,0 +1,221 @@
+package metrics
+
+// Ported from
+// https://github.com/dropwizard/metrics/blob/release/4.2.x/metrics-core/src/main/java/com/codahale/metrics/ChunkedAssociativeLongArray.java
+
+import (
+	"github.com/gammazero/deque"
+	"sort"
+	"strconv"
+	"strings"
+)
+
+const (
+	ChunkedAssociativeArrayDefaultChunkSize = 512
+	ChunkedAssociativeArrayMaxCacheSize     = 128
+)
+
+type ChunkedAssociativeArray struct {
+	defaultChunkSize int
+
+	/*
+	 * We use this ArrayDeque as cache to store chunks that are expired and removed from main data structure.
+	 * Then instead of allocating new AssociativeArrayChunk immediately we are trying to poll one from this deque.
+	 * So if you have constant or slowly changing load ChunkedAssociativeLongArray will never
+	 * throw away old chunks or allocate new ones which makes this data structure almost garbage free.
+	 */
+	chunksCache *deque.Deque[*AssociativeArrayChunk]
+	chunks      *deque.Deque[*AssociativeArrayChunk]
+}
+
+func NewChunkedAssociativeArray(chunkSize int) *ChunkedAssociativeArray {
+	return &ChunkedAssociativeArray{
+		defaultChunkSize: chunkSize,
+		chunksCache:      deque.New[*AssociativeArrayChunk](ChunkedAssociativeArrayMaxCacheSize, ChunkedAssociativeArrayMaxCacheSize),
+		chunks:           deque.New[*AssociativeArrayChunk](),
+	}
+}
+
+func (caa *ChunkedAssociativeArray) Clear() {
+	for i := 0; i < caa.chunks.Len(); i++ {
+		chunk := caa.chunks.PopBack()
+		caa.freeChunk(chunk)
+	}
+}
+
+func (caa *ChunkedAssociativeArray) AllocateChunk() *AssociativeArrayChunk {
+	if caa.chunksCache.Len() == 0 {
+		return NewAssociativeArrayChunk(caa.defaultChunkSize)
+	}
+
+	chunk := caa.chunksCache.PopBack()
+	chunk.cursor = 0
+	chunk.startIndex = 0
+	chunk.chunkSize = len(chunk.keys)
+
+	return chunk
+}
+
+func (caa *ChunkedAssociativeArray) freeChunk(chunk *AssociativeArrayChunk) {
+	if caa.chunksCache.Len() < ChunkedAssociativeArrayMaxCacheSize {
+		caa.chunksCache.PushBack(chunk)
+	}
+}
+
+func (caa *ChunkedAssociativeArray) Put(key int64, value int64) {
+	var activeChunk *AssociativeArrayChunk
+	if caa.chunks.Len() > 0 {
+		activeChunk = caa.chunks.Back()
+	}
+
+	if activeChunk != nil && activeChunk.cursor != 0 && activeChunk.keys[activeChunk.cursor-1] > key {
+		// Key must be the same as last inserted or bigger
+		key = activeChunk.keys[activeChunk.cursor-1] + 1
+	}
+	if activeChunk == nil || activeChunk.cursor-activeChunk.startIndex == activeChunk.chunkSize {
+		// The last chunk doesn't exist or full
+		activeChunk = caa.AllocateChunk()
+		caa.chunks.PushBack(activeChunk)
+	}
+	activeChunk.Append(key, value)
+}
+
+func (caa *ChunkedAssociativeArray) Values() []int64 {
+	valuesSize := caa.Size()
+	if valuesSize == 0 {
+		// Empty
+		return []int64{0}
+	}
+
+	values := make([]int64, 0, valuesSize)
+	caa.chunks.Index(func(chunk *AssociativeArrayChunk) bool {
+		values = append(values, chunk.values[chunk.startIndex:chunk.cursor]...)
+		return false
+	})
+
+	return values
+}
+
+func (caa *ChunkedAssociativeArray) Size() int {
+	var result int
+	caa.chunks.Index(func(chunk *AssociativeArrayChunk) bool {
+		result += chunk.cursor - chunk.startIndex
+		return false
+	})
+	return result
+}
+
+func (caa *ChunkedAssociativeArray) String() string {
+	var builder strings.Builder
+	first := true
+	caa.chunks.Index(func(chunk *AssociativeArrayChunk) bool {
+		if first {
+			first = false
+		} else {
+			builder.WriteString("->")
+		}
+		builder.WriteString("[")
+		for i := chunk.startIndex; i < chunk.cursor; i++ {
+			builder.WriteString("(")
+			builder.WriteString(strconv.FormatInt(chunk.keys[i], 10))
+			builder.WriteString(": ")
+			builder.WriteString(strconv.FormatInt(chunk.values[i], 10))
+			builder.WriteString(") ")
+		}
+		builder.WriteString("]")
+		return false
+	})
+
+	return builder.String()
+}
+
+// Trim tries to trim all beyond specified boundaries
+// startKey: the start value for which all elements less than it should be removed.
+// endKey:   the end value for which all elements greater/equals than it should be removed
+func (caa *ChunkedAssociativeArray) Trim(startKey int64, endKey int64) {
+	/*
+	 * [3, 4, 5, 9] -> [10, 13, 14, 15] -> [21, 24, 29, 30] -> [31] :: start layout
+	 *       |5______________________________23|                    :: trim(5, 23)
+	 *       [5, 9] -> [10, 13, 14, 15] -> [21]                     :: result layout
+	 */
+	// Remove elements that are too large
+	indexBeforeEndKey := caa.chunks.RIndex(func(chunk *AssociativeArrayChunk) bool {
+		if chunk.IsFirstElementEmptyOrGreaterEqualThanKey(endKey) {
+			return false
+		}
+
+		chunk.cursor = chunk.FindFirstIndexOfGreaterEqualElements(endKey)
+		return true
+	})
+
+	// Remove chunks that only contain elements that are too large
+	if indexBeforeEndKey >= 0 {
+		for i := caa.chunks.Len() - 1; i > indexBeforeEndKey; i-- {
+			chunk := caa.chunks.PopBack()
+			caa.freeChunk(chunk)
+		}
+	}
+
+	// Remove elements that are too small
+	indexAfterStartKey := caa.chunks.Index(func(chunk *AssociativeArrayChunk) bool {
+		if chunk.IsLastElementEmptyOrLessThanKey(startKey) {
+			return false
+		}
+
+		newStartIndex := chunk.FindFirstIndexOfGreaterEqualElements(startKey)
+		if chunk.startIndex != newStartIndex {
+			chunk.startIndex = newStartIndex
+			chunk.chunkSize = chunk.cursor - chunk.startIndex
+		}
+		return true
+	})
+
+	// Remove chunks that only contain elements that are too small
+	for i := 0; i < indexAfterStartKey; i++ {
+		chunk := caa.chunks.PopFront()
+		caa.freeChunk(chunk)
+	}
+}
+
+type AssociativeArrayChunk struct {
+	keys   []int64
+	values []int64
+
+	chunkSize  int
+	startIndex int
+	cursor     int
+}
+
+func NewAssociativeArrayChunk(chunkSize int) *AssociativeArrayChunk {
+	return &AssociativeArrayChunk{
+		keys:       make([]int64, chunkSize),
+		values:     make([]int64, chunkSize),
+		chunkSize:  chunkSize,
+		startIndex: 0,
+		cursor:     0,
+	}
+}
+
+func (c *AssociativeArrayChunk) Append(key int64, value int64) {
+	c.keys[c.cursor] = key
+	c.values[c.cursor] = value
+	c.cursor++
+}
+
+func (c *AssociativeArrayChunk) IsFirstElementEmptyOrGreaterEqualThanKey(key int64) bool {
+	return c.cursor == c.startIndex || c.keys[c.startIndex] >= key
+}
+
+func (c *AssociativeArrayChunk) IsLastElementEmptyOrLessThanKey(key int64) bool {
+	return c.cursor == c.startIndex || c.keys[c.cursor-1] < key
+}
+
+func (c *AssociativeArrayChunk) FindFirstIndexOfGreaterEqualElements(minKey int64) int {
+	if c.cursor == c.startIndex || c.keys[c.startIndex] >= minKey {
+		return c.startIndex
+	}
+	elements := c.keys[c.startIndex:c.cursor]
+	keyIndex := sort.Search(len(elements), func(i int) bool { return elements[i] >= minKey })
+
+	return c.startIndex + keyIndex
+}