Rule Concurrency: Simpler loop for sequential (default) executions ()

This commit is contained in:
Julien Duchesne 2025-01-09 18:14:21 -05:00 committed by GitHub
parent f030894c2c
commit 0a19f1268e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 36 additions and 27 deletions

View file

@ -643,28 +643,46 @@ func (g *Group) Eval(ctx context.Context, ts time.Time) {
if ctrl == nil {
ctrl = sequentialRuleEvalController{}
}
for _, batch := range ctrl.SplitGroupIntoBatches(ctx, g) {
for _, ruleIndex := range batch {
batches := ctrl.SplitGroupIntoBatches(ctx, g)
if len(batches) == 0 {
// Sequential evaluation when batches aren't set.
// This is the behaviour without a defined RuleConcurrencyController
for i, rule := range g.rules {
// Check if the group has been stopped.
select {
case <-g.done:
return
default:
}
rule := g.rules[ruleIndex]
if len(batch) > 1 && ctrl.Allow(ctx, g, rule) {
wg.Add(1)
go eval(ruleIndex, rule, func() {
wg.Done()
ctrl.Done(ctx)
})
} else {
eval(ruleIndex, rule, nil)
}
eval(i, rule, nil)
}
} else {
// Concurrent evaluation.
for _, batch := range batches {
for _, ruleIndex := range batch {
// Check if the group has been stopped.
select {
case <-g.done:
wg.Wait()
return
default:
}
rule := g.rules[ruleIndex]
if len(batch) > 1 && ctrl.Allow(ctx, g, rule) {
wg.Add(1)
go eval(ruleIndex, rule, func() {
wg.Done()
ctrl.Done(ctx)
})
} else {
eval(ruleIndex, rule, nil)
}
}
// It is important that we finish processing any rules in this current batch - before we move into the next one.
wg.Wait()
}
// It is important that we finish processing any rules in this current batch - before we move into the next one.
wg.Wait()
}
g.metrics.GroupSamples.WithLabelValues(GroupKey(g.File(), g.Name())).Set(samplesTotal.Load())

View file

@ -550,11 +550,7 @@ func (c sequentialRuleEvalController) Allow(_ context.Context, _ *Group, _ Rule)
}
func (c sequentialRuleEvalController) SplitGroupIntoBatches(_ context.Context, g *Group) []ConcurrentRules {
order := make([]ConcurrentRules, len(g.rules))
for i := range g.rules {
order[i] = []int{i}
}
return order
return nil
}
func (c sequentialRuleEvalController) Done(_ context.Context) {}

View file

@ -1989,12 +1989,7 @@ func TestAsyncRuleEvaluation(t *testing.T) {
// Expected evaluation order
order := group.opts.RuleConcurrencyController.SplitGroupIntoBatches(ctx, group)
require.Equal(t, []ConcurrentRules{
{0},
{1},
{2},
{3},
}, order)
require.Nil(t, order)
// Never expect more than 1 inflight query at a time.
require.EqualValues(t, 1, maxInflight.Load())