Skip to content

Commit 9f09a9a

Browse files
GiedriusSGiedriusS
authored
store/bucket: snappy-encoded postings reading improvements (#6245)
* store: pool input to snappy.Decode Pool input to snappy.Decode to avoid allocations. Signed-off-by: Giedrius Statkevičius <[email protected]> * store: use s2 for decoding snappy It's faster hence use it. Signed-off-by: Giedrius Statkevičius <[email protected]> * store: small code style adjustment Signed-off-by: Giedrius Statkevičius <[email protected]> * store: call closefns before returning err Signed-off-by: Giedrius Statkevičius <[email protected]> * store/postings_codec: return both if possible Signed-off-by: Giedrius Statkevičius <[email protected]> * store/bucket: always call close fns Signed-off-by: Giedrius Statkevičius <[email protected]> --------- Signed-off-by: Giedrius Statkevičius <[email protected]>
1 parent 5d5d39a commit 9f09a9a

File tree

3 files changed

+62
-17
lines changed

3 files changed

+62
-17
lines changed

‎pkg/store/bucket.go‎

Lines changed: 18 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2162,7 +2162,12 @@ func (r *bucketIndexReader) ExpandedPostings(ctx context.Context, ms []*labels.M
21622162
keys = append(keys, allPostingsLabel)
21632163
}
21642164

2165-
fetchedPostings, err := r.fetchPostings(ctx, keys, bytesLimiter)
2165+
fetchedPostings, closeFns, err := r.fetchPostings(ctx, keys, bytesLimiter)
2166+
defer func() {
2167+
for _, closeFn := range closeFns {
2168+
closeFn()
2169+
}
2170+
}()
21662171
if err != nil {
21672172
return nil, errors.Wrap(err, "get postings")
21682173
}
@@ -2302,7 +2307,9 @@ type postingPtr struct {
23022307
// fetchPostings fill postings requested by posting groups.
23032308
// It returns one postings for each key, in the same order.
23042309
// If postings for given key is not fetched, entry at given index will be nil.
2305-
func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Label, bytesLimiter BytesLimiter) ([]index.Postings, error) {
2310+
func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Label, bytesLimiter BytesLimiter) ([]index.Postings, []func(), error) {
2311+
var closeFns []func()
2312+
23062313
timer := prometheus.NewTimer(r.block.metrics.postingsFetchDuration)
23072314
defer timer.ObserveDuration()
23082315

@@ -2314,7 +2321,7 @@ func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Lab
23142321
fromCache, _ := r.block.indexCache.FetchMultiPostings(ctx, r.block.meta.ULID, keys)
23152322
for _, dataFromCache := range fromCache {
23162323
if err := bytesLimiter.Reserve(uint64(len(dataFromCache))); err != nil {
2317-
return nil, errors.Wrap(err, "bytes limit exceeded while loading postings from index cache")
2324+
return nil, closeFns, errors.Wrap(err, "bytes limit exceeded while loading postings from index cache")
23182325
}
23192326
}
23202327

@@ -2335,18 +2342,21 @@ func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Lab
23352342
)
23362343
if isDiffVarintSnappyEncodedPostings(b) {
23372344
s := time.Now()
2338-
l, err = diffVarintSnappyDecode(b)
2345+
clPostings, err := diffVarintSnappyDecode(b)
23392346
r.stats.cachedPostingsDecompressions += 1
23402347
r.stats.CachedPostingsDecompressionTimeSum += time.Since(s)
23412348
if err != nil {
23422349
r.stats.cachedPostingsDecompressionErrors += 1
2350+
} else {
2351+
closeFns = append(closeFns, clPostings.close)
2352+
l = clPostings
23432353
}
23442354
} else {
23452355
_, l, err = r.dec.Postings(b)
23462356
}
23472357

23482358
if err != nil {
2349-
return nil, errors.Wrap(err, "decode postings")
2359+
return nil, closeFns, errors.Wrap(err, "decode postings")
23502360
}
23512361

23522362
output[ix] = l
@@ -2362,7 +2372,7 @@ func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Lab
23622372
}
23632373

23642374
if err != nil {
2365-
return nil, errors.Wrap(err, "index header PostingsOffset")
2375+
return nil, closeFns, errors.Wrap(err, "index header PostingsOffset")
23662376
}
23672377

23682378
r.stats.postingsToFetch++
@@ -2384,7 +2394,7 @@ func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Lab
23842394
length := int64(part.End) - start
23852395

23862396
if err := bytesLimiter.Reserve(uint64(length)); err != nil {
2387-
return nil, errors.Wrap(err, "bytes limit exceeded while fetching postings")
2397+
return nil, closeFns, errors.Wrap(err, "bytes limit exceeded while fetching postings")
23882398
}
23892399
}
23902400

@@ -2462,7 +2472,7 @@ func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Lab
24622472
})
24632473
}
24642474

2465-
return output, g.Wait()
2475+
return output, closeFns, g.Wait()
24662476
}
24672477

24682478
func resizePostings(b []byte) ([]byte, error) {

‎pkg/store/postings_codec.go‎

Lines changed: 42 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -5,8 +5,10 @@ package store
55

66
import (
77
"bytes"
8+
"sync"
89

910
"github.com/golang/snappy"
11+
"github.com/klauspost/compress/s2"
1012
"github.com/pkg/errors"
1113
"github.com/prometheus/prometheus/storage"
1214
"github.com/prometheus/prometheus/tsdb/encoding"
@@ -82,27 +84,60 @@ func diffVarintEncodeNoHeader(p index.Postings, length int) ([]byte, error) {
8284
return buf.B, nil
8385
}
8486

85-
func diffVarintSnappyDecode(input []byte) (index.Postings, error) {
87+
var snappyDecodePool sync.Pool
88+
89+
type closeablePostings interface {
90+
index.Postings
91+
close()
92+
}
93+
94+
// alias returns true if given slices have the same both backing array.
95+
// See: https://groups.google.com/g/golang-nuts/c/C6ufGl73Uzk.
96+
func alias(x, y []byte) bool {
97+
return cap(x) > 0 && cap(y) > 0 && &x[0:cap(x)][cap(x)-1] == &y[0:cap(y)][cap(y)-1]
98+
}
99+
100+
func diffVarintSnappyDecode(input []byte) (closeablePostings, error) {
86101
if !isDiffVarintSnappyEncodedPostings(input) {
87102
return nil, errors.New("header not found")
88103
}
89104

90-
raw, err := snappy.Decode(nil, input[len(codecHeaderSnappy):])
105+
toFree := make([][]byte, 0, 2)
106+
107+
var dstBuf []byte
108+
decodeBuf := snappyDecodePool.Get()
109+
if decodeBuf != nil {
110+
dstBuf = *(decodeBuf.(*[]byte))
111+
toFree = append(toFree, dstBuf)
112+
}
113+
114+
raw, err := s2.Decode(dstBuf, input[len(codecHeaderSnappy):])
91115
if err != nil {
92116
return nil, errors.Wrap(err, "snappy decode")
93117
}
94118

95-
return newDiffVarintPostings(raw), nil
119+
if !alias(raw, dstBuf) {
120+
toFree = append(toFree, raw)
121+
}
122+
123+
return newDiffVarintPostings(raw, toFree), nil
96124
}
97125

98-
func newDiffVarintPostings(input []byte) *diffVarintPostings {
99-
return &diffVarintPostings{buf: &encoding.Decbuf{B: input}}
126+
func newDiffVarintPostings(input []byte, freeSlices [][]byte) *diffVarintPostings {
127+
return &diffVarintPostings{freeSlices: freeSlices, buf: &encoding.Decbuf{B: input}}
100128
}
101129

102130
// diffVarintPostings is an implementation of index.Postings based on diff+varint encoded data.
103131
type diffVarintPostings struct {
104-
buf *encoding.Decbuf
105-
cur storage.SeriesRef
132+
buf *encoding.Decbuf
133+
cur storage.SeriesRef
134+
freeSlices [][]byte
135+
}
136+
137+
func (it *diffVarintPostings) close() {
138+
for i := range it.freeSlices {
139+
snappyDecodePool.Put(&it.freeSlices[i])
140+
}
106141
}
107142

108143
func (it *diffVarintPostings) At() storage.SeriesRef {

‎pkg/store/postings_codec_test.go‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -55,9 +55,9 @@ func TestDiffVarintCodec(t *testing.T) {
5555

5656
codecs := map[string]struct {
5757
codingFunction func(index.Postings, int) ([]byte, error)
58-
decodingFunction func([]byte) (index.Postings, error)
58+
decodingFunction func([]byte) (closeablePostings, error)
5959
}{
60-
"raw": {codingFunction: diffVarintEncodeNoHeader, decodingFunction: func(bytes []byte) (index.Postings, error) { return newDiffVarintPostings(bytes), nil }},
60+
"raw": {codingFunction: diffVarintEncodeNoHeader, decodingFunction: func(bytes []byte) (closeablePostings, error) { return newDiffVarintPostings(bytes, nil), nil }},
6161
"snappy": {codingFunction: diffVarintSnappyEncode, decodingFunction: diffVarintSnappyDecode},
6262
}
6363

0 commit comments

Comments
 (0)