Skip to content

Commit 6204558

Browse files
SungJin1212SungJin1212
committed
Add prealloc timeseries v2
Signed-off-by: SungJin1212 <[email protected]>
1 parent 6778f49 commit 6204558

File tree

5 files changed

+259
-12
lines changed

5 files changed

+259
-12
lines changed

‎pkg/cortexpb/slicesPool.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ type byteSlicePools struct {
1313
pools []sync.Pool
1414
}
1515

16-
func newSlicePool(pools int) *byteSlicePools {
16+
func NewSlicePool(pools int) *byteSlicePools {
1717
sp := byteSlicePools{}
1818
sp.init(pools)
1919
return &sp
@@ -32,7 +32,7 @@ func (sp *byteSlicePools) init(pools int) {
3232
}
3333
}
3434

35-
func (sp *byteSlicePools) getSlice(size int) *[]byte {
35+
func (sp *byteSlicePools) GetSlice(size int) *[]byte {
3636
index := int(math.Ceil(math.Log2(float64(size)))) - minPoolSizePower
3737

3838
if index >= len(sp.pools) {
@@ -50,7 +50,7 @@ func (sp *byteSlicePools) getSlice(size int) *[]byte {
5050
return s
5151
}
5252

53-
func (sp *byteSlicePools) reuseSlice(s *[]byte) {
53+
func (sp *byteSlicePools) ReuseSlice(s *[]byte) {
5454
index := int(math.Floor(math.Log2(float64(cap(*s))))) - minPoolSizePower
5555

5656
if index >= len(sp.pools) || index < 0 {

‎pkg/cortexpb/slicesPool_test.go

+6-6
Original file line numberDiff line numberDiff line change
@@ -9,22 +9,22 @@ import (
99
)
1010

1111
func TestFuzzyByteSlicePools(t *testing.T) {
12-
sut := newSlicePool(20)
12+
sut := NewSlicePool(20)
1313
maxByteSize := int(math.Pow(2, 20+minPoolSizePower-1))
1414

1515
for i := 0; i < 1000; i++ {
1616
size := rand.Int() % maxByteSize
17-
s := sut.getSlice(size)
17+
s := sut.GetSlice(size)
1818
assert.Equal(t, len(*s), size)
19-
sut.reuseSlice(s)
19+
sut.ReuseSlice(s)
2020
}
2121
}
2222

2323
func TestReturnSliceSmallerThanMin(t *testing.T) {
24-
sut := newSlicePool(20)
24+
sut := NewSlicePool(20)
2525
size := 3
2626
buff := make([]byte, 0, size)
27-
sut.reuseSlice(&buff)
28-
buff2 := sut.getSlice(size * 2)
27+
sut.ReuseSlice(&buff)
28+
buff2 := sut.GetSlice(size * 2)
2929
assert.Equal(t, len(*buff2), size*2)
3030
}

‎pkg/cortexpb/timeseries.go

+3-3
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,7 @@ var (
4747
}
4848
},
4949
}
50-
bytePool = newSlicePool(20)
50+
bytePool = NewSlicePool(20)
5151
)
5252

5353
// PreallocConfig configures how structures will be preallocated to optimise
@@ -86,7 +86,7 @@ func (p *PreallocTimeseries) Unmarshal(dAtA []byte) error {
8686

8787
func (p *PreallocWriteRequest) Marshal() (dAtA []byte, err error) {
8888
size := p.Size()
89-
p.data = bytePool.getSlice(size)
89+
p.data = bytePool.GetSlice(size)
9090
dAtA = *p.data
9191
n, err := p.MarshalToSizedBuffer(dAtA[:size])
9292
if err != nil {
@@ -97,7 +97,7 @@ func (p *PreallocWriteRequest) Marshal() (dAtA []byte, err error) {
9797

9898
func ReuseWriteRequest(req *PreallocWriteRequest) {
9999
if req.data != nil {
100-
bytePool.reuseSlice(req.data)
100+
bytePool.ReuseSlice(req.data)
101101
req.data = nil
102102
}
103103
req.Source = 0

‎pkg/cortexpbv2/timeseriesv2.go

+136
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
package cortexpbv2
2+
3+
import (
4+
"sync"
5+
6+
"github.com/cortexproject/cortex/pkg/cortexpb"
7+
)
8+
9+
var (
10+
expectedTimeseries = 100
11+
expectedLabels = 20
12+
expectedSymbols = 20
13+
expectedSamplesPerSeries = 10
14+
expectedExemplarsPerSeries = 1
15+
expectedHistogramsPerSeries = 1
16+
17+
slicePool = sync.Pool{
18+
New: func() interface{} {
19+
return make([]PreallocTimeseriesV2, 0, expectedTimeseries)
20+
},
21+
}
22+
23+
timeSeriesPool = sync.Pool{
24+
New: func() interface{} {
25+
return &TimeSeries{
26+
LabelsRefs: make([]uint32, 0, expectedLabels),
27+
Samples: make([]Sample, 0, expectedSamplesPerSeries),
28+
Histograms: make([]Histogram, 0, expectedHistogramsPerSeries),
29+
Exemplars: make([]Exemplar, 0, expectedExemplarsPerSeries),
30+
Metadata: Metadata{},
31+
}
32+
},
33+
}
34+
35+
writeRequestPool = sync.Pool{
36+
New: func() interface{} {
37+
return &PreallocWriteRequestV2{
38+
WriteRequest: WriteRequest{
39+
Symbols: make([]string, 0, expectedSymbols),
40+
},
41+
}
42+
},
43+
}
44+
bytePool = cortexpb.NewSlicePool(20)
45+
)
46+
47+
// PreallocWriteRequestV2 is a WriteRequest which preallocs slices on Unmarshal.
48+
type PreallocWriteRequestV2 struct {
49+
WriteRequest
50+
data *[]byte
51+
}
52+
53+
// Unmarshal implements proto.Message.
54+
func (p *PreallocWriteRequestV2) Unmarshal(dAtA []byte) error {
55+
p.Timeseries = PreallocTimeseriesV2SliceFromPool()
56+
return p.WriteRequest.Unmarshal(dAtA)
57+
}
58+
59+
func (p *PreallocWriteRequestV2) Marshal() (dAtA []byte, err error) {
60+
size := p.Size()
61+
p.data = bytePool.GetSlice(size)
62+
dAtA = *p.data
63+
n, err := p.MarshalToSizedBuffer(dAtA[:size])
64+
if err != nil {
65+
return nil, err
66+
}
67+
return dAtA[:n], nil
68+
}
69+
70+
// PreallocTimeseriesV2 is a TimeSeries which preallocs slices on Unmarshal.
71+
type PreallocTimeseriesV2 struct {
72+
*TimeSeries
73+
}
74+
75+
// Unmarshal implements proto.Message.
76+
func (p *PreallocTimeseriesV2) Unmarshal(dAtA []byte) error {
77+
p.TimeSeries = TimeseriesV2FromPool()
78+
return p.TimeSeries.Unmarshal(dAtA)
79+
}
80+
81+
func ReuseWriteRequestV2(req *PreallocWriteRequestV2) {
82+
if req.data != nil {
83+
bytePool.ReuseSlice(req.data)
84+
req.data = nil
85+
}
86+
req.Source = 0
87+
req.Symbols = nil
88+
req.Timeseries = nil
89+
writeRequestPool.Put(req)
90+
}
91+
92+
func PreallocWriteRequestV2FromPool() *PreallocWriteRequestV2 {
93+
return writeRequestPool.Get().(*PreallocWriteRequestV2)
94+
}
95+
96+
// PreallocTimeseriesV2SliceFromPool retrieves a slice of PreallocTimeseriesV2 from a sync.Pool.
97+
// ReuseSlice should be called once done.
98+
func PreallocTimeseriesV2SliceFromPool() []PreallocTimeseriesV2 {
99+
return slicePool.Get().([]PreallocTimeseriesV2)
100+
}
101+
102+
// ReuseSlice puts the slice back into a sync.Pool for reuse.
103+
func ReuseSlice(ts []PreallocTimeseriesV2) {
104+
for i := range ts {
105+
ReuseTimeseries(ts[i].TimeSeries)
106+
}
107+
108+
slicePool.Put(ts[:0]) //nolint:staticcheck //see comment on slicePool for more details
109+
}
110+
111+
// TimeseriesV2FromPool retrieves a pointer to a TimeSeries from a sync.Pool.
112+
// ReuseTimeseries should be called once done, unless ReuseSlice was called on the slice that contains this TimeSeries.
113+
func TimeseriesV2FromPool() *TimeSeries {
114+
return timeSeriesPool.Get().(*TimeSeries)
115+
}
116+
117+
// ReuseTimeseries puts the timeseries back into a sync.Pool for reuse.
118+
func ReuseTimeseries(ts *TimeSeries) {
119+
// clear ts lableRef and samples
120+
ts.LabelsRefs = ts.LabelsRefs[:0]
121+
ts.Samples = ts.Samples[:0]
122+
123+
// clear exmplar labelrefs
124+
for i := range ts.Exemplars {
125+
ts.Exemplars[i].LabelsRefs = ts.Exemplars[i].LabelsRefs[:0]
126+
}
127+
128+
for i := range ts.Histograms {
129+
ts.Histograms[i].Reset()
130+
}
131+
132+
ts.Exemplars = ts.Exemplars[:0]
133+
ts.Histograms = ts.Histograms[:0]
134+
ts.Metadata = Metadata{}
135+
timeSeriesPool.Put(ts)
136+
}

‎pkg/cortexpbv2/timeseriesv2_test.go

+111
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,111 @@
1+
package cortexpbv2
2+
3+
import (
4+
"fmt"
5+
"testing"
6+
7+
"github.com/gogo/protobuf/proto"
8+
"github.com/stretchr/testify/assert"
9+
"github.com/stretchr/testify/require"
10+
)
11+
12+
func TestPreallocTimeseriesV2SliceFromPool(t *testing.T) {
13+
t.Run("new instance is provided when not available to reuse", func(t *testing.T) {
14+
first := PreallocTimeseriesV2SliceFromPool()
15+
second := PreallocTimeseriesV2SliceFromPool()
16+
17+
assert.NotSame(t, first, second)
18+
})
19+
20+
t.Run("instance is cleaned before reusing", func(t *testing.T) {
21+
slice := PreallocTimeseriesV2SliceFromPool()
22+
slice = append(slice, PreallocTimeseriesV2{TimeSeries: &TimeSeries{}})
23+
ReuseSlice(slice)
24+
25+
reused := PreallocTimeseriesV2SliceFromPool()
26+
assert.Len(t, reused, 0)
27+
})
28+
}
29+
30+
func TestTimeseriesV2FromPool(t *testing.T) {
31+
t.Run("new instance is provided when not available to reuse", func(t *testing.T) {
32+
first := TimeseriesV2FromPool()
33+
second := TimeseriesV2FromPool()
34+
35+
assert.NotSame(t, first, second)
36+
})
37+
38+
t.Run("instance is cleaned before reusing", func(t *testing.T) {
39+
ts := TimeseriesV2FromPool()
40+
ts.LabelsRefs = []uint32{1, 2}
41+
ts.Samples = []Sample{{Value: 1, Timestamp: 2}}
42+
ts.Exemplars = []Exemplar{{LabelsRefs: []uint32{1, 2}, Value: 1, Timestamp: 2}}
43+
ts.Histograms = []Histogram{{}}
44+
fmt.Println("ts.Histograms", len(ts.Histograms))
45+
ReuseTimeseries(ts)
46+
47+
reused := TimeseriesV2FromPool()
48+
assert.Len(t, reused.LabelsRefs, 0)
49+
assert.Len(t, reused.Samples, 0)
50+
assert.Len(t, reused.Exemplars, 0)
51+
assert.Len(t, reused.Histograms, 0)
52+
})
53+
}
54+
55+
func BenchmarkMarshallWriteRequest(b *testing.B) {
56+
ts := PreallocTimeseriesV2SliceFromPool()
57+
58+
for i := 0; i < 100; i++ {
59+
ts = append(ts, PreallocTimeseriesV2{TimeSeries: TimeseriesV2FromPool()})
60+
ts[i].LabelsRefs = []uint32{1, 2, 3, 4, 5, 6, 7, 8}
61+
ts[i].Samples = []Sample{{Value: 1, Timestamp: 2}}
62+
}
63+
64+
tests := []struct {
65+
name string
66+
writeRequestFactory func() proto.Marshaler
67+
clean func(in interface{})
68+
}{
69+
{
70+
name: "no-pool",
71+
writeRequestFactory: func() proto.Marshaler {
72+
return &WriteRequest{Timeseries: ts}
73+
},
74+
clean: func(in interface{}) {},
75+
},
76+
{
77+
name: "byte pool",
78+
writeRequestFactory: func() proto.Marshaler {
79+
w := &PreallocWriteRequestV2{}
80+
w.Timeseries = ts
81+
return w
82+
},
83+
clean: func(in interface{}) {
84+
ReuseWriteRequestV2(in.(*PreallocWriteRequestV2))
85+
},
86+
},
87+
{
88+
name: "byte and write pool",
89+
writeRequestFactory: func() proto.Marshaler {
90+
w := PreallocWriteRequestV2FromPool()
91+
w.Timeseries = ts
92+
return w
93+
},
94+
clean: func(in interface{}) {
95+
ReuseWriteRequestV2(in.(*PreallocWriteRequestV2))
96+
},
97+
},
98+
}
99+
100+
for _, tc := range tests {
101+
b.Run(tc.name, func(b *testing.B) {
102+
for i := 0; i < b.N; i++ {
103+
w := tc.writeRequestFactory()
104+
_, err := w.Marshal()
105+
require.NoError(b, err)
106+
tc.clean(w)
107+
}
108+
b.ReportAllocs()
109+
})
110+
}
111+
}

0 commit comments

Comments
 (0)