fix(dispatch): reduce locking contention

Reduce the amount of time spent holding locks in the dispatcher:

- doMaintenance() holds RW lock only during deletion
- Groups() holds R lock, copies, releases
- processAlert() holds R(W) locks only when necessary

This results in -70% maintenance overhead or +19518.99% alert processing rate improvement:
```
                                                 │ bench-dispatch-main.txt │        bench-dispatch-fix-locking.txt         │
                                                 │   baseline_alerts/sec   │ baseline_alerts/sec  vs base                  │
Dispatch_100k_AggregationGroups_10k_Empty-12                 1.899M ±   1%          1.919M ±  2%          ~ (p=0.063 n=10)
Dispatch_100k_AggregationGroups_20k_Empty-12                 1.934M ±   2%          1.967M ±  1%     +1.69% (p=0.015 n=10)
Dispatch_100k_AggregationGroups_30k_Empty-12                 1.926M ±   3%          1.931M ±  4%          ~ (p=0.436 n=10)
Dispatch_100k_AggregationGroups_40k_Empty-12                 2.087M ±  10%          2.030M ±  7%          ~ (p=0.912 n=10)
Dispatch_100k_AggregationGroups_50k_Empty-12                 1.922M ±   3%          2.118M ± 10%    +10.16% (p=0.000 n=10)
Dispatch_20k_AggregationGroups_Groups_Impact-12              180.7k ±  42%         2128.6k ±  4%  +1077.93% (p=0.000 n=10)
Dispatch_50k_AggregationGroups_Groups_Impact-12              57.02k ± 196%        2089.55k ±  2%  +3564.69% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_Groups_Impact-12             19.61k ±  35%        1899.27k ±  2%  +9582.74% (p=0.000 n=10)
geomean                                                      524.6k                 2.008M         +282.84%

                                             │ bench-dispatch-main.txt │         bench-dispatch-fix-locking.txt         │
                                             │ maintenance_overhead_%  │ maintenance_overhead_%  vs base                │
Dispatch_100k_AggregationGroups_10k_Empty-12              17.185 ±  7%              5.672 ± 45%  -66.99% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_20k_Empty-12               36.50 ± 11%              12.56 ± 14%  -65.59% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_30k_Empty-12               55.44 ± 13%              23.11 ± 30%  -58.32% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_40k_Empty-12              125.65 ± 27%              28.98 ± 53%  -76.94% (p=0.000 n=10)
Dispatch_100k_AggregationGroups_50k_Empty-12              172.40 ± 36%              37.76 ± 82%  -78.10% (p=0.000 n=10)
geomean                                                    59.62                    17.83        -70.10%
```

Signed-off-by: Siavash Safi <[email protected]>

Author: siavashs

dispatch/dispatch.go

@@ -19,6 +19,7 @@ import (
 	"log/slog"
 	"sort"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/prometheus/common/model"
@@ -41,8 +42,7 @@ type Dispatcher struct {
 	timeout func(time.Duration) time.Duration
 
 	mtx                sync.RWMutex
-	aggrGroupsPerRoute map[*Route]map[model.Fingerprint]*aggrGroup
-	aggrGroupsNum      int
+	aggrGroupsPerRoute routeGroups
 
 	done   chan struct{}
 	ctx    context.Context
@@ -84,8 +84,10 @@ func (d *Dispatcher) Run() {
 	d.done = make(chan struct{})
 
 	d.mtx.Lock()
-	d.aggrGroupsPerRoute = map[*Route]map[model.Fingerprint]*aggrGroup{}
-	d.aggrGroupsNum = 0
+	d.aggrGroupsPerRoute = routeGroups{
+		groupsNum: &atomic.Int64{},
+		limits:    d.limits,
+	}
 	d.metrics.aggrGroups.Set(0)
 	d.ctx, d.cancel = context.WithCancel(context.Background())
 	d.mtx.Unlock()
@@ -134,40 +136,58 @@ func (d *Dispatcher) run(it provider.AlertIterator) {
 }
 
 func (d *Dispatcher) doMaintenance() {
-	d.mtx.Lock()
-	defer d.mtx.Unlock()
-	for _, groups := range d.aggrGroupsPerRoute {
-		for _, ag := range groups {
+	type groupToRemove struct {
+		route *Route
+		fp    model.Fingerprint
+		ag    *aggrGroup
+	}
+
+	var toRemove []groupToRemove
+
+	// First pass: collect groups to remove
+	d.aggrGroupsPerRoute.Range(func(route *Route, groups *fingerprintGroups) bool {
+		groups.Range(func(fp model.Fingerprint, ag *aggrGroup) bool {
 			if ag.empty() {
-				ag.stop()
-				d.marker.DeleteByGroupKey(ag.routeID, ag.GroupKey())
-				delete(groups, ag.fingerprint())
-				d.aggrGroupsNum--
-				d.metrics.aggrGroups.Dec()
+				toRemove = append(toRemove, groupToRemove{route, fp, ag})
 			}
+			return true
+		})
+		return true
+	})
+
+	// Second pass: remove collected groups
+	for _, item := range toRemove {
+		item.ag.stop()
+		d.marker.DeleteByGroupKey(item.ag.routeID, item.ag.GroupKey())
+		groupsMap := d.aggrGroupsPerRoute.getRoute(item.route)
+		if groupsMap != nil {
+			groupsMap.removeGroup(item.fp)
+			d.metrics.aggrGroups.Dec()
 		}
 	}
 }
 
 // Groups returns a slice of AlertGroups from the dispatcher's internal state.
 func (d *Dispatcher) Groups(routeFilter func(*Route) bool, alertFilter func(*types.Alert, time.Time) bool) (AlertGroups, map[model.Fingerprint][]string) {
-	groups := AlertGroups{}
-
-	d.mtx.RLock()
-	defer d.mtx.RUnlock()
+	// Snapshot the outer map in routeGroups to
+	//  avoid holding the read lock when dispatcher has 1000s of aggregation groups.
+	routeGroups := routeGroups{}
+	d.aggrGroupsPerRoute.Range(func(route *Route, groups *fingerprintGroups) bool {
+		routeGroups.addRoute(route)
+		return true
+	})
 
-	// Keep a list of receivers for an alert to prevent checking each alert
-	// again against all routes. The alert has already matched against this
-	// route on ingestion.
+	// TODO: move this processing out of Dispatcher, it does not belong here.
+	alertGroups := AlertGroups{}
 	receivers := map[model.Fingerprint][]string{}
-
 	now := time.Now()
-	for route, ags := range d.aggrGroupsPerRoute {
+	routeGroups.Range(func(route *Route, _ *fingerprintGroups) bool {
 		if !routeFilter(route) {
-			continue
+			return true
 		}
 
-		for _, ag := range ags {
+		// Read inner fingerprintGroups if necessary.
+		d.aggrGroupsPerRoute.getRoute(route).Range(func(fp model.Fingerprint, ag *aggrGroup) bool {
 			receiver := route.RouteOpts.Receiver
 			alertGroup := &AlertGroup{
 				Labels:   ag.labels,
@@ -197,22 +217,24 @@ func (d *Dispatcher) Groups(routeFilter func(*Route) bool, alertFilter func(*typ
 				filteredAlerts = append(filteredAlerts, a)
 			}
 			if len(filteredAlerts) == 0 {
-				continue
+				return true
 			}
 			alertGroup.Alerts = filteredAlerts
 
-			groups = append(groups, alertGroup)
-		}
-	}
-	sort.Sort(groups)
-	for i := range groups {
-		sort.Sort(groups[i].Alerts)
+			alertGroups = append(alertGroups, alertGroup)
+			return true
+		})
+		return true
+	})
+	sort.Sort(alertGroups)
+	for i := range alertGroups {
+		sort.Sort(alertGroups[i].Alerts)
 	}
 	for i := range receivers {
 		sort.Strings(receivers[i])
 	}
 
-	return groups, receivers
+	return alertGroups, receivers
 }
 
 // Stop the dispatcher.
@@ -236,42 +258,31 @@ func (d *Dispatcher) Stop() {
 // and inserts it.
 func (d *Dispatcher) processAlert(alert *types.Alert, route *Route) {
 	groupLabels := getGroupLabels(alert, route)
-
 	fp := groupLabels.Fingerprint()
 
-	d.mtx.Lock()
-	defer d.mtx.Unlock()
-
-	routeGroups, ok := d.aggrGroupsPerRoute[route]
-	if !ok {
-		routeGroups = map[model.Fingerprint]*aggrGroup{}
-		d.aggrGroupsPerRoute[route] = routeGroups
-	}
-
-	ag, ok := routeGroups[fp]
-	if ok {
-		ag.insert(alert)
+	routeGroups := d.aggrGroupsPerRoute.addRoute(route)
+	group := routeGroups.getGroup(fp)
+	if group != nil {
+		group.insert(alert)
 		return
 	}
 
-	// If the group does not exist, create it. But check the limit first.
-	if limit := d.limits.MaxNumberOfAggregationGroups(); limit > 0 && d.aggrGroupsNum >= limit {
+	// If the group does not exist, create it.
+	group, count, limit := routeGroups.addGroup(fp, newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.logger))
+	if group == nil {
+		// Rate limited.
 		d.metrics.aggrGroupLimitReached.Inc()
-		d.logger.Error("Too many aggregation groups, cannot create new group for alert", "groups", d.aggrGroupsNum, "limit", limit, "alert", alert.Name())
+		d.logger.Error("Too many aggregation groups, cannot create new group for alert", "groups", count, "limit", limit)
 		return
 	}
-
-	ag = newAggrGroup(d.ctx, groupLabels, route, d.timeout, d.logger)
-	routeGroups[fp] = ag
-	d.aggrGroupsNum++
 	d.metrics.aggrGroups.Inc()
 
 	// Insert the 1st alert in the group before starting the group's run()
 	// function, to make sure that when the run() will be executed the 1st
 	// alert is already there.
-	ag.insert(alert)
+	group.insert(alert)
 
-	go ag.run(func(ctx context.Context, alerts ...*types.Alert) bool {
+	go group.run(func(ctx context.Context, alerts ...*types.Alert) bool {
 		_, _, err := d.stage.Exec(ctx, d.logger, alerts...)
 		if err != nil {
 			logger := d.logger.With("num_alerts", len(alerts), "err", err)