Merge 00e202f693 into 57d2b552c7

eth/protocols/eth/handler.go

@@ -197,9 +197,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

@@ -26,6 +26,7 @@ require (
 	github.com/fjl/gencodec v0.0.0-20230517082657-f9840df7b83e
 	github.com/fjl/memsize v0.0.2
 	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
@@ -99,7 +100,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=
@@ -198,6 +199,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=
@@ -206,6 +209,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=

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
+}

metrics/chunked_associative_array_test.go

@@ -0,0 +1,170 @@
+package metrics
+
+import (
+	"testing"
+)
+
+// Ported from
+// https://github.com/dropwizard/metrics/blob/release/4.2.x/metrics-core/src/main/java/com/codahale/metrics/ChunkedAssociativeLongArray.java
+
+func TestChunkedAssociativeArray_Put(t *testing.T) {
+	array := NewChunkedAssociativeArray(3)
+	// Test that time cannot go backwards
+	array.Put(7, 7)
+	expectedStringBefore := "[(7: 7) ]"
+	if array.String() != expectedStringBefore {
+		t.Errorf("initial array string incorrect: %s", array.String())
+	}
+}
+
+func TestChunkedAssociativeArray_ValuesEmpty(t *testing.T) {
+	array := NewChunkedAssociativeArray(3)
+
+	values := array.Values()
+	if len(values) != 1 {
+		t.Fatalf("unexpected length of values: %d", len(values))
+	}
+	if values[0] != 0 {
+		t.Errorf("unexpected value in empty values: %d", values[0])
+	}
+}
+
+func TestChunkedAssociativeArray_Trim(t *testing.T) {
+	array := NewChunkedAssociativeArray(3)
+	array.Put(-7, 7)
+	array.Put(-5, 7)
+	array.Put(-4, 7)
+	array.Put(-3, 3)
+	array.Put(-2, 1)
+	array.Put(0, 5)
+	array.Put(3, 0)
+	array.Put(9, 8)
+	array.Put(15, 0)
+	array.Put(19, 5)
+	array.Put(21, 5)
+	array.Put(34, -9)
+	array.Put(109, 5)
+
+	expectedStringBefore := "[(-7: 7) (-5: 7) (-4: 7) ]->[(-3: 3) (-2: 1) (0: 5) ]->[(3: 0) (9: 8) (15: 0) ]->[(19: 5) (21: 5) (34: -9) ]->[(109: 5) ]"
+	if array.String() != expectedStringBefore {
+		t.Errorf("initial array string incorrect: %s", array.String())
+	}
+	valuesBefore := array.Values()
+	expectedValuesBefore := []int64{7, 7, 7, 3, 1, 5, 0, 8, 0, 5, 5, -9, 5}
+	if len(valuesBefore) != len(expectedValuesBefore) {
+		t.Errorf("initial values returned incorrect length: %d", len(valuesBefore))
+	} else {
+		for i, value := range valuesBefore {
+			if value != expectedValuesBefore[i] {
+				t.Errorf("unexpected value %d at index %d", value, i)
+			}
+		}
+	}
+	if array.Size() != 13 {
+		t.Errorf("initial array size incorrect: %d", array.Size())
+	}
+
+	array.Trim(-2, 20)
+
+	expectedStringAfter := "[(-2: 1) (0: 5) ]->[(3: 0) (9: 8) (15: 0) ]->[(19: 5) ]"
+	if array.String() != expectedStringAfter {
+		t.Errorf("array string incorrect: %s", array.String())
+	}
+	valuesAfter := array.Values()
+	expectedValuesAfter := []int64{1, 5, 0, 8, 0, 5}
+	if len(valuesAfter) != len(expectedValuesAfter) {
+		t.Errorf("values returned incorrect length: %d", len(valuesAfter))
+	} else {
+		for i, value := range valuesAfter {
+			if value != expectedValuesAfter[i] {
+				t.Errorf("unexpected value %d at index %d", value, i)
+			}
+		}
+	}
+	if array.Size() != 6 {
+		t.Errorf("array size incorrect: %d", array.Size())
+	}
+
+	array.Trim(-2, 16)
+	expectedStringAfter2 := "[(-2: 1) (0: 5) ]->[(3: 0) (9: 8) (15: 0) ]"
+	if array.String() != expectedStringAfter2 {
+		t.Errorf("array string incorrect: %s", array.String())
+	}
+
+	initialCacheCount := array.chunksCache.Len()
+
+	// Have AllocateChunk take from cache
+	array.Put(200, 555)
+	expectedStringAfter3 := "[(-2: 1) (0: 5) ]->[(3: 0) (9: 8) (15: 0) ]->[(200: 555) ]"
+	if array.String() != expectedStringAfter3 {
+		t.Errorf("array string incorrect: %s", array.String())
+	}
+
+	if array.chunksCache.Len() >= initialCacheCount {
+		t.Error("cache not used when allocating chunk")
+	}
+}
+
+func TestAssociativeArrayChunk_IsFirstElementEmptyOrGreaterEqualThanKey(t *testing.T) {
+	chunk := NewAssociativeArrayChunk(3)
+
+	if !chunk.IsFirstElementEmptyOrGreaterEqualThanKey(5) {
+		t.Error("empty test failed")
+	}
+
+	chunk.keys = []int64{41, 42, 43}
+	chunk.startIndex = 1
+
+	if chunk.IsFirstElementEmptyOrGreaterEqualThanKey(43) {
+		t.Error("element less than key test failed")
+	}
+	if !chunk.IsFirstElementEmptyOrGreaterEqualThanKey(42) {
+		t.Error("element greater than or equal to key test failed")
+	}
+}
+
+func TestAssociativeArrayChunk_IsLastElementIsLessThanKey(t *testing.T) {
+	chunk := NewAssociativeArrayChunk(3)
+
+	if !chunk.IsLastElementEmptyOrLessThanKey(5) {
+		t.Error("empty test failed")
+	}
+
+	chunk.keys = []int64{41, 42, 43}
+	chunk.cursor = 2
+
+	if !chunk.IsLastElementEmptyOrLessThanKey(43) {
+		t.Error("element less than key test failed")
+	}
+	if chunk.IsLastElementEmptyOrLessThanKey(42) {
+		t.Error("element greater than or equal to key test failed")
+	}
+}
+
+func TestAssociativeArrayChunk_FindFirstIndexOfGreaterEqualElements(t *testing.T) {
+	chunk := NewAssociativeArrayChunk(7)
+
+	if chunk.FindFirstIndexOfGreaterEqualElements(5) != 0 {
+		t.Error("empty test failed")
+	}
+
+	chunk.keys = []int64{41, 42, 43, 44, 45, 46, 48}
+	chunk.startIndex = 1
+	chunk.cursor = 6
+
+	if chunk.FindFirstIndexOfGreaterEqualElements(41) != chunk.startIndex {
+		t.Error("minKey less than first element test failed")
+	}
+	if chunk.FindFirstIndexOfGreaterEqualElements(42) != chunk.startIndex {
+		t.Error("minKey greater than or equal to first element test failed")
+	}
+	if chunk.FindFirstIndexOfGreaterEqualElements(43) != 2 {
+		t.Error("2nd element test failed")
+	}
+	if chunk.FindFirstIndexOfGreaterEqualElements(46) != 5 {
+		t.Error("last element test failed")
+	}
+	if chunk.FindFirstIndexOfGreaterEqualElements(49) != 6 {
+		t.Error("past last element test failed")
+	}
+}

metrics/sample.go

@@ -32,6 +32,10 @@ type Sample interface {
 	Update(int64)
 }
 
+func NewBoundedHistogramSample() Sample {
+	return NewSlidingTimeWindowArraySample(time.Minute * 1)
+}
+
 // ExpDecaySample is an exponentially-decaying sample using a forward-decaying
 // priority reservoir.  See Cormode et al's "Forward Decay: A Practical Time
 // Decay Model for Streaming Systems".

metrics/sample_test.go

@@ -87,6 +87,18 @@ func BenchmarkUniformSample1028(b *testing.B) {
 	benchmarkSample(b, NewUniformSample(1028))
 }
 
+func BenchmarkSlidingWindowArraySample257(b *testing.B) {
+	benchmarkSample(b, NewSlidingTimeWindowArraySample(257))
+}
+
+func BenchmarkSlidingWindowArraySample514(b *testing.B) {
+	benchmarkSample(b, NewSlidingTimeWindowArraySample(514))
+}
+
+func BenchmarkSlidingWindowArraySample1028(b *testing.B) {
+	benchmarkSample(b, NewSlidingTimeWindowArraySample(1028))
+}
+
 func min(a, b int) int {
 	if a < b {
 		return a

metrics/sliding_time_window_array_sample.go

@@ -0,0 +1,123 @@
+package metrics
+
+import (
+	"math"
+	"sync"
+	"time"
+)
+
+// SlidingTimeWindowArraySample is ported from Coda Hale's dropwizard library
+// <https://github.com/dropwizard/metrics/pull/1139>
+// A reservoir implementation backed by a sliding window that stores only the
+// measurements made in the last given window of time
+type SlidingTimeWindowArraySample struct {
+	startTick    int64
+	measurements *ChunkedAssociativeArray
+	window       int64
+	count        int64
+	lastTick     int64
+	mutex        sync.Mutex
+}
+
+const (
+	// SlidingTimeWindowCollisionBuffer allow this many duplicate ticks
+	// before overwriting measurements
+	SlidingTimeWindowCollisionBuffer = 256
+
+	// SlidingTimeWindowTrimThreshold is number of updates between trimming data
+	SlidingTimeWindowTrimThreshold = 256
+
+	// SlidingTimeWindowClearBufferTicks is the number of ticks to keep past the
+	// requested trim
+	SlidingTimeWindowClearBufferTicks = int64(time.Hour/time.Nanosecond) *
+		SlidingTimeWindowCollisionBuffer
+)
+
+// NewSlidingTimeWindowArraySample creates new object with given window of time
+func NewSlidingTimeWindowArraySample(window time.Duration) Sample {
+	if !Enabled {
+		return NilSample{}
+	}
+	return &SlidingTimeWindowArraySample{
+		startTick:    time.Now().UnixNano(),
+		measurements: NewChunkedAssociativeArray(ChunkedAssociativeArrayDefaultChunkSize),
+		window:       window.Nanoseconds() * SlidingTimeWindowCollisionBuffer,
+	}
+}
+
+// Clear clears all samples.
+func (s *SlidingTimeWindowArraySample) Clear() {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	s.count = 0
+	s.measurements.Clear()
+}
+
+// trim requires s.mutex to already be acquired
+func (s *SlidingTimeWindowArraySample) trim() {
+	now := s.getTick()
+	windowStart := now - s.window
+	windowEnd := now + SlidingTimeWindowClearBufferTicks
+	if windowStart < windowEnd {
+		s.measurements.Trim(windowStart, windowEnd)
+	} else {
+		// long overflow handling that can only happen 1 year after class loading
+		s.measurements.Clear()
+	}
+}
+
+// getTick requires s.mutex to already be acquired
+func (s *SlidingTimeWindowArraySample) getTick() int64 {
+	oldTick := s.lastTick
+	tick := (time.Now().UnixNano() - s.startTick) * SlidingTimeWindowCollisionBuffer
+	var newTick int64
+	if tick-oldTick > 0 {
+		newTick = tick
+	} else {
+		newTick = oldTick + 1
+	}
+	s.lastTick = newTick
+	return newTick
+}
+
+// Snapshot returns a read-only copy of the sample.
+func (s *SlidingTimeWindowArraySample) Snapshot() SampleSnapshot {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	s.trim()
+	var (
+		samples       = s.measurements.Values()
+		values        = make([]int64, len(samples))
+		max     int64 = math.MinInt64
+		min     int64 = math.MaxInt64
+		sum     int64
+	)
+	for i, v := range samples {
+		values[i] = v
+		sum += v
+		if v > max {
+			max = v
+		}
+		if v < min {
+			min = v
+		}
+	}
+	return newSampleSnapshotPrecalculated(s.count, values, min, max, sum)
+}
+
+// Update samples a new value.
+func (s *SlidingTimeWindowArraySample) Update(v int64) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+	var newTick int64
+	s.count += 1
+	if s.count%SlidingTimeWindowTrimThreshold == 0 {
+		s.trim()
+	}
+	newTick = s.getTick()
+	longOverflow := newTick < s.lastTick
+	if longOverflow {
+		s.measurements.Clear()
+	}
+	s.measurements.Put(newTick, v)
+}

p2p/tracker/tracker.go

@@ -197,9 +197,7 @@ func (t *Tracker) Fulfil(peer string, version uint, code uint64, id uint64) {
 
 	h := fmt.Sprintf("%s/%s/%d/%#02x", waitHistName, t.protocol, req.version, req.reqCode)
 	sampler := func() metrics.Sample {
-		return metrics.ResettingSample(
-			metrics.NewExpDecaySample(1028, 0.015),
-		)
+		return metrics.NewBoundedHistogramSample()
 	}
 	metrics.GetOrRegisterHistogramLazy(h, nil, sampler).Update(time.Since(req.time).Microseconds())
 }

rpc/metrics.go

@@ -42,9 +42,7 @@ func updateServeTimeHistogram(method string, success bool, elapsed time.Duration
 	}
 	h := fmt.Sprintf("%s/%s/%s", serveTimeHistName, method, note)
 	sampler := func() metrics.Sample {
-		return metrics.ResettingSample(
-			metrics.NewExpDecaySample(1028, 0.015),
-		)
+		return metrics.NewBoundedHistogramSample()
 	}
 	metrics.GetOrRegisterHistogramLazy(h, nil, sampler).Update(elapsed.Nanoseconds())
 }