store/bucket: snappy-encoded postings reading improvements (thanos-io…

…#6245) * store: pool input to snappy.Decode Pool input to snappy.Decode to avoid allocations. Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * store: use s2 for decoding snappy It's faster hence use it. Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * store: small code style adjustment Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * store: call closefns before returning err Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * store/postings_codec: return both if possible Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> * store/bucket: always call close fns Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com> --------- Signed-off-by: Giedrius Statkevičius <giedrius.statkevicius@vinted.com>
vinted · Apr 17, 2023 · e8971c6 · e8971c6
1 parent b982d4f
commit e8971c6
Show file tree

Hide file tree

Showing 4 changed files with 24 additions and 16 deletions.
diff --git a/go.mod b/go.mod
@@ -126,7 +126,7 @@ require (
 	golang.org/x/exp v0.0.0-20230307190834-24139beb5833
 )
 
-require go4.org/unsafe/assume-no-moving-gc v0.0.0-20220617031537-928513b29760 // indirect
+require go4.org/unsafe/assume-no-moving-gc v0.0.0-20230221090011-e4bae7ad2296 // indirect
 
 require (
 	cloud.google.com/go/compute/metadata v0.2.3 // indirect

diff --git a/go.sum b/go.sum
@@ -1030,8 +1030,9 @@ go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
 go.uber.org/zap v1.13.0/go.mod h1:zwrFLgMcdUuIBviXEYEH1YKNaOBnKXsx2IPda5bBwHM=
 go4.org/intern v0.0.0-20220617035311-6925f38cc365 h1:t9hFvR102YlOqU0fQn1wgwhNvSbHGBbbJxX9JKfU3l0=
 go4.org/intern v0.0.0-20220617035311-6925f38cc365/go.mod h1:WXRv3p7T6gzt0CcJm43AAKdKVZmcQbwwC7EwquU5BZU=
-go4.org/unsafe/assume-no-moving-gc v0.0.0-20220617031537-928513b29760 h1:FyBZqvoA/jbNzuAWLQE2kG820zMAkcilx6BMjGbL/E4=
 go4.org/unsafe/assume-no-moving-gc v0.0.0-20220617031537-928513b29760/go.mod h1:FftLjUGFEDu5k8lt0ddY+HcrH/qU/0qk+H8j9/nTl3E=
+go4.org/unsafe/assume-no-moving-gc v0.0.0-20230221090011-e4bae7ad2296 h1:QJ/xcIANMLApehfgPCHnfK1hZiaMmbaTVmPv7DAoTbo=
+go4.org/unsafe/assume-no-moving-gc v0.0.0-20230221090011-e4bae7ad2296/go.mod h1:FftLjUGFEDu5k8lt0ddY+HcrH/qU/0qk+H8j9/nTl3E=
 golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
 golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=

diff --git a/pkg/store/bucket.go b/pkg/store/bucket.go
@@ -2370,7 +2370,7 @@ func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Lab
 		}
 
 		if err != nil {
-			return nil, nil, errors.Wrap(err, "index header PostingsOffset")
+			return nil, closeFns, errors.Wrap(err, "index header PostingsOffset")
 		}
 
 		r.stats.postingsToFetch++
@@ -2392,7 +2392,7 @@ func (r *bucketIndexReader) fetchPostings(ctx context.Context, keys []labels.Lab
 		length := int64(part.End) - start
 
 		if err := bytesLimiter.Reserve(uint64(length)); err != nil {
-			return nil, nil, errors.Wrap(err, "bytes limit exceeded while fetching postings")
+			return nil, closeFns, errors.Wrap(err, "bytes limit exceeded while fetching postings")
 		}
 	}
 

diff --git a/pkg/store/postings_codec.go b/pkg/store/postings_codec.go
@@ -91,46 +91,53 @@ type closeablePostings interface {
 	close()
 }
 
+// alias returns true if given slices have the same both backing array.
+// See: https://groups.google.com/g/golang-nuts/c/C6ufGl73Uzk.
+func alias(x, y []byte) bool {
+	return cap(x) > 0 && cap(y) > 0 && &x[0:cap(x)][cap(x)-1] == &y[0:cap(y)][cap(y)-1]
+}
+
 func diffVarintSnappyDecode(input []byte) (closeablePostings, error) {
 	if !isDiffVarintSnappyEncodedPostings(input) {
 		return nil, errors.New("header not found")
 	}
 
+	toFree := make([][]byte, 0, 2)
+
 	var dstBuf []byte
 	decodeBuf := snappyDecodePool.Get()
 	if decodeBuf != nil {
 		dstBuf = *(decodeBuf.(*[]byte))
+		toFree = append(toFree, dstBuf)
 	}
 
 	raw, err := s2.Decode(dstBuf, input[len(codecHeaderSnappy):])
 	if err != nil {
 		return nil, errors.Wrap(err, "snappy decode")
 	}
 
-	biggerSlice := raw
-	if cap(dstBuf) > cap(biggerSlice) {
-		biggerSlice = dstBuf
+	if !alias(raw, dstBuf) {
+		toFree = append(toFree, raw)
 	}
 
-	return newDiffVarintPostings(raw, biggerSlice), nil
+	return newDiffVarintPostings(raw, toFree), nil
 }
 
-func newDiffVarintPostings(input, freeSlice []byte) *diffVarintPostings {
-	return &diffVarintPostings{freeSlice: freeSlice, buf: &encoding.Decbuf{B: input}}
+func newDiffVarintPostings(input []byte, freeSlices [][]byte) *diffVarintPostings {
+	return &diffVarintPostings{freeSlices: freeSlices, buf: &encoding.Decbuf{B: input}}
 }
 
 // diffVarintPostings is an implementation of index.Postings based on diff+varint encoded data.
 type diffVarintPostings struct {
-	buf       *encoding.Decbuf
-	cur       storage.SeriesRef
-	freeSlice []byte
+	buf        *encoding.Decbuf
+	cur        storage.SeriesRef
+	freeSlices [][]byte
 }
 
 func (it *diffVarintPostings) close() {
-	if it.freeSlice == nil {
-		return
+	for i := range it.freeSlices {
+		snappyDecodePool.Put(&it.freeSlices[i])
 	}
-	snappyDecodePool.Put(&it.freeSlice)
 }
 
 func (it *diffVarintPostings) At() storage.SeriesRef {