mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-07 06:53:09 -06:00
Code Issues Packages Projects Releases Wiki Activity

This probably makes more sense

Browse source
This commit is contained in:
Till Faelligen 2023-11-01 12:27:47 +01:00
parent 116b0b96e2
commit be0c6995e9
No known key found for this signature in database
GPG key ID: ACCDC9606D472758
1 changed files with 3 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
userapi/internal/device_list_update.go
View file

@ -367,8 +367,6 @@ func (u *DeviceListUpdater) notifyWorkers(userID string) {
index := int(int64(hash.Sum32()) % int64(len(u.workerChans))) index := int(int64(hash.Sum32()) % int64(len(u.workerChans)))
ch := u.assignChannel(userID) ch := u.assignChannel(userID)
deviceListUpdaterBackpressure.With(prometheus.Labels{"worker_id": strconv.Itoa(index)}).Inc()
defer deviceListUpdaterBackpressure.With(prometheus.Labels{"worker_id": strconv.Itoa(index)}).Dec()
u.workerChans[index] <- remoteServer u.workerChans[index] <- remoteServer
select { select {
case <-ch: case <-ch:
@ -420,17 +418,17 @@ func (u *DeviceListUpdater) worker(ch chan spec.ServerName, workerID int) {
deviceListUpdaterServersRetrying.With(prometheus.Labels{"worker_id": strconv.Itoa(workerID)}).Set(float64(len(retries))) deviceListUpdaterServersRetrying.With(prometheus.Labels{"worker_id": strconv.Itoa(workerID)}).Set(float64(len(retries)))
retriesMu.Unlock() retriesMu.Unlock()
for _, srv := range serversToRetry { for _, srv := range serversToRetry {
deviceListUpdaterBackpressure.With(prometheus.Labels{"worker_id": strconv.Itoa(workerID)}).Inc()
ch <- srv ch <- srv
deviceListUpdaterBackpressure.With(prometheus.Labels{"worker_id": strconv.Itoa(workerID)}).Dec()
} }
} }
}() }()
for serverName := range ch { for serverName := range ch {
deviceListUpdaterBackpressure.With(prometheus.Labels{"worker_id": strconv.Itoa(workerID)}).Inc()
retriesMu.Lock() retriesMu.Lock()
_, exists := retries[serverName] _, exists := retries[serverName]
retriesMu.Unlock() retriesMu.Unlock()
if exists { if exists {
deviceListUpdaterBackpressure.With(prometheus.Labels{"worker_id": strconv.Itoa(workerID)}).Dec()
// Don't retry a server that we're already waiting for. // Don't retry a server that we're already waiting for.
continue continue
} }
@ -442,6 +440,7 @@ func (u *DeviceListUpdater) worker(ch chan spec.ServerName, workerID int) {
} }
retriesMu.Unlock() retriesMu.Unlock()
} }
deviceListUpdaterBackpressure.With(prometheus.Labels{"worker_id": strconv.Itoa(workerID)}).Dec()
} }
} }