mudhorn/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2025-02-02 08:31:11 -08:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

Refactor PostingsForMatcherCache promise

Browse source 
Extract promise payload as a struct, to make size calculation easier.

Signed-off-by: Oleg Zaytsev <mail@olegzaytsev.com>
This commit is contained in:
Oleg Zaytsev 2023-09-27 17:23:03 +02:00
parent 241a342b33
commit 5fdf784243
2 changed files with 25 additions and 23 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
tsdb/postings_for_matchers_cache.go
View file

@ -81,41 +81,43 @@ func (c *PostingsForMatchersCache) PostingsForMatchers(ctx context.Context, ix I
return c.postingsForMatchersPromise(ctx, ix, ms)() return c.postingsForMatchersPromise(ctx, ix, ms)()
} }
func (c *PostingsForMatchersCache) postingsForMatchersPromise(ctx context.Context, ix IndexPostingsReader, ms []*labels.Matcher) func() (index.Postings, error) { type postingsForMatcherPromise struct {
var ( sync.WaitGroup
wg sync.WaitGroup
cloner *index.PostingsCloner
outerErr error
)
wg.Add(1)
promise := func() (index.Postings, error) { cloner *index.PostingsCloner
wg.Wait() err error
if outerErr != nil { }
return nil, outerErr
} func (p *postingsForMatcherPromise) result() (index.Postings, error) {
return cloner.Clone(), nil p.Wait()
if p.err != nil {
return nil, p.err
} }
return p.cloner.Clone(), nil
}
func (c *PostingsForMatchersCache) postingsForMatchersPromise(ctx context.Context, ix IndexPostingsReader, ms []*labels.Matcher) func() (index.Postings, error) {
promise := new(postingsForMatcherPromise)
promise.Add(1)
key := matchersKey(ms) key := matchersKey(ms)
oldPromise, loaded := c.calls.LoadOrStore(key, promise) oldPromise, loaded := c.calls.LoadOrStore(key, promise)
if loaded { if loaded {
return oldPromise.(func() (index.Postings, error)) promise = oldPromise.(*postingsForMatcherPromise)
return promise.result
} }
defer wg.Done() defer promise.Done()
if postings, err := c.postingsForMatchers(ctx, ix, ms...); err != nil { if postings, err := c.postingsForMatchers(ctx, ix, ms...); err != nil {
outerErr = err promise.err = err
} else { } else {
cloner = index.NewPostingsCloner(postings) promise.cloner = index.NewPostingsCloner(postings)
} }
// Estimate the size of the cache entry, in bytes. We use max() because sizeBytes := int64(len(key) + size.Of(promise))
// size.Of() returns -1 if the value is nil.
sizeBytes := int64(len(key)) + max(0, int64(size.Of(outerErr))) + max(0, int64(size.Of(cloner)))
c.created(key, c.timeNow(), sizeBytes) c.created(key, c.timeNow(), sizeBytes)
return promise return promise.result
} }
type postingsForMatchersCachedCall struct { type postingsForMatchersCachedCall struct {

4
tsdb/postings_for_matchers_cache_test.go
View file

@ -267,7 +267,7 @@ func TestPostingsForMatchersCache(t *testing.T) {
t.Run("cached value is evicted because cache exceeds max bytes", func(t *testing.T) { t.Run("cached value is evicted because cache exceeds max bytes", func(t *testing.T) {
const ( const (
maxItems = 100 // Never hit it. maxItems = 100 // Never hit it.
maxBytes = 1000 maxBytes = 1100
numMatchers = 5 numMatchers = 5
postingsListSize = 30 // 8 bytes per posting ref, so 30 x 8 = 240 bytes. postingsListSize = 30 // 8 bytes per posting ref, so 30 x 8 = 240 bytes.
) )
@ -313,7 +313,7 @@ func TestPostingsForMatchersCache(t *testing.T) {
// At this point we expect that the postings have been computed only once for the 3 matchers. // At this point we expect that the postings have been computed only once for the 3 matchers.
for i := 0; i < 3; i++ { for i := 0; i < 3; i++ {
assert.Equal(t, 1, callsPerMatchers[matchersKey(matchersLists[i])]) assert.Equalf(t, 1, callsPerMatchers[matchersKey(matchersLists[i])], "matcher %d", i)
} }
// Call PostingsForMatchers() for a 4th matcher. We expect this will evict the oldest cached entry. // Call PostingsForMatchers() for a 4th matcher. We expect this will evict the oldest cached entry.