Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Fix flakes in ./topdown/cache #7188

Open
wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Fix flakes in ./topdown/cache #7188

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions topdown/cache/cache.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,16 +154,14 @@ func NewInterQueryCache(config *Config) InterQueryCache {
func NewInterQueryCacheWithContext(ctx context.Context, config *Config) InterQueryCache {
iqCache := newCache(config)
if iqCache.staleEntryEvictionTimePeriodSeconds() > 0 {
cleanupTicker := time.NewTicker(time.Duration(iqCache.staleEntryEvictionTimePeriodSeconds()) * time.Second)
go func() {
cleanupTicker := time.NewTicker(time.Duration(iqCache.staleEntryEvictionTimePeriodSeconds()) * time.Second)
defer cleanupTicker.Stop()
for {
select {
case <-cleanupTicker.C:
cleanupTicker.Stop()
iqCache.cleanStaleValues()
cleanupTicker = time.NewTicker(time.Duration(iqCache.staleEntryEvictionTimePeriodSeconds()) * time.Second)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Are we ok with this in scenarios where cleanStaleValues routine can consume more time than staleEntryEvictionTimePeriodSeconds?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can switch this back, but since the lock is held throughout cleanStaleValues, the actual evaluation experience will certainly be suffering either way.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That should be fine IMO. I am more concerned if the cleanup routine keeps running all the time when ticker is faster than time taken to scan all the cache entries. Earlier the ticker was reset after a cleanup routine was done that ensured a gap of staleEntryEvictionTimePeriodSeconds between cleanup runs.

case <-ctx.Done():
cleanupTicker.Stop()
return
}
}
Expand Down
32 changes: 22 additions & 10 deletions topdown/cache/cache_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -408,14 +408,17 @@ func TestInsertWithExpiryAndEviction(t *testing.T) {
t.Fatalf("Unexpected error %v", err)
}

cache := NewInterQueryCacheWithContext(context.Background(), config)
// This starts a background ticker at stale_entry_eviction_period_seconds to clean up items.
ctx, cancel := context.WithCancel(context.Background())
cache := NewInterQueryCacheWithContext(ctx, config)
t.Cleanup(cancel)

cacheValue := newInterQueryCacheValue(ast.StringTerm("bar").Value, 20)
cache.InsertWithExpiry(ast.StringTerm("force_evicted_foo").Value, cacheValue, time.Now().Add(100*time.Second))
if fetchedCacheValue, found := cache.Get(ast.StringTerm("force_evicted_foo").Value); !found {
t.Fatalf("Expected cache entry with value %v, found %v", cacheValue, fetchedCacheValue)
}
cache.InsertWithExpiry(ast.StringTerm("expired_foo").Value, cacheValue, time.Now().Add(1*time.Second))
cache.InsertWithExpiry(ast.StringTerm("expired_foo").Value, cacheValue, time.Now().Add(900*time.Millisecond))
if fetchedCacheValue, found := cache.Get(ast.StringTerm("expired_foo").Value); !found {
t.Fatalf("Expected cache entry with value %v, found %v", cacheValue, fetchedCacheValue)
}
Expand All @@ -425,7 +428,7 @@ func TestInsertWithExpiryAndEviction(t *testing.T) {
}

// Ensure stale entries clean up routine runs at least once
time.Sleep(2 * time.Second)
time.Sleep(1100 * time.Millisecond)

// Entry deleted even though not expired because force evicted when foo is inserted
if fetchedCacheValue, found := cache.Get(ast.StringTerm("force_evicted_foo").Value); found {
Expand Down Expand Up @@ -454,20 +457,23 @@ func TestInsertHighTTLWithStaleEntryCleanup(t *testing.T) {
t.Fatalf("Unexpected error %v", err)
}

cache := NewInterQueryCacheWithContext(context.Background(), config)
// This starts a background ticker at stale_entry_eviction_period_seconds to clean up items.
ctx, cancel := context.WithCancel(context.Background())
cache := NewInterQueryCacheWithContext(ctx, config)
t.Cleanup(cancel)

cacheValue := newInterQueryCacheValue(ast.StringTerm("bar").Value, 20)
cache.InsertWithExpiry(ast.StringTerm("high_ttl_foo").Value, cacheValue, time.Now().Add(100*time.Second))
if fetchedCacheValue, found := cache.Get(ast.StringTerm("high_ttl_foo").Value); !found {
t.Fatalf("Expected cache entry with value %v, found %v", cacheValue, fetchedCacheValue)
}
cache.InsertWithExpiry(ast.StringTerm("expired_foo").Value, cacheValue, time.Now().Add(1*time.Second))
cache.InsertWithExpiry(ast.StringTerm("expired_foo").Value, cacheValue, time.Now().Add(900*time.Millisecond))
if fetchedCacheValue, found := cache.Get(ast.StringTerm("expired_foo").Value); !found {
t.Fatalf("Expected cache entry with value %v, found no entry", fetchedCacheValue)
}

// Ensure stale entries clean up routine runs at least once
time.Sleep(2 * time.Second)
time.Sleep(1100 * time.Millisecond)

cache.InsertWithExpiry(ast.StringTerm("foo").Value, cacheValue, time.Now().Add(10*time.Second))
if fetchedCacheValue, found := cache.Get(ast.StringTerm("foo").Value); !found {
Expand Down Expand Up @@ -497,7 +503,10 @@ func TestInsertHighTTLWithoutStaleEntryCleanup(t *testing.T) {
t.Fatalf("Unexpected error %v", err)
}

cache := NewInterQueryCacheWithContext(context.Background(), config)
// This starts a background ticker at stale_entry_eviction_period_seconds to clean up items.
ctx, cancel := context.WithCancel(context.Background())
cache := NewInterQueryCacheWithContext(ctx, config)
t.Cleanup(cancel)

cacheValue := newInterQueryCacheValue(ast.StringTerm("bar").Value, 20)
cache.InsertWithExpiry(ast.StringTerm("high_ttl_foo").Value, cacheValue, time.Now().Add(100*time.Second))
Expand Down Expand Up @@ -537,14 +546,17 @@ func TestZeroExpiryTime(t *testing.T) {
t.Fatalf("Unexpected error %v", err)
}

cache := NewInterQueryCacheWithContext(context.Background(), config)
// This starts a background ticker at stale_entry_eviction_period_seconds to clean up items.
ctx, cancel := context.WithCancel(context.Background())
cache := NewInterQueryCacheWithContext(ctx, config)
t.Cleanup(cancel)
cacheValue := newInterQueryCacheValue(ast.StringTerm("bar").Value, 20)
cache.InsertWithExpiry(ast.StringTerm("foo").Value, cacheValue, time.Time{})
if fetchedCacheValue, found := cache.Get(ast.StringTerm("foo").Value); !found {
t.Fatalf("Expected cache entry with value %v for foo, found %v", cacheValue, fetchedCacheValue)
}

time.Sleep(2 * time.Second)
time.Sleep(1100 * time.Millisecond)

// Stale entry cleanup routine skips zero time cache entries
if fetchedCacheValue, found := cache.Get(ast.StringTerm("foo").Value); !found {
Expand Down Expand Up @@ -574,7 +586,7 @@ func TestCancelNewInterQueryCacheWithContext(t *testing.T) {
}

cancel()
time.Sleep(2 * time.Second)
time.Sleep(1100 * time.Millisecond)

// Stale entry cleanup routine stopped as context was cancelled
if fetchedCacheValue, found := cache.Get(ast.StringTerm("foo").Value); !found {
Expand Down