mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2026-01-01 03:03:10 -06:00
Code Issues Packages Projects Releases Wiki Activity

Wake destination queues gradually, rather than all at once

Browse source
This commit is contained in:
Neil Alexander 2022-04-04 13:17:08 +01:00
parent 6748a2a823
commit be24eea797
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
1 changed files with 19 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
federationapi/queue/queue.go
View file

@ -104,28 +104,28 @@ func NewOutgoingQueues(
} }
// Look up which servers we have pending items for and then rehydrate those queues. // Look up which servers we have pending items for and then rehydrate those queues.
if !disabled { if !disabled {
time.AfterFunc(time.Second*5, func() { serverNames := map[gomatrixserverlib.ServerName]struct{}{}
serverNames := map[gomatrixserverlib.ServerName]struct{}{} if names, err := db.GetPendingPDUServerNames(context.Background()); err == nil {
if names, err := db.GetPendingPDUServerNames(context.Background()); err == nil { for _, serverName := range names {
for _, serverName := range names { serverNames[serverName] = struct{}{}
serverNames[serverName] = struct{}{}
}
} else {
log.WithError(err).Error("Failed to get PDU server names for destination queue hydration")
} }
if names, err := db.GetPendingEDUServerNames(context.Background()); err == nil { } else {
for _, serverName := range names { log.WithError(err).Error("Failed to get PDU server names for destination queue hydration")
serverNames[serverName] = struct{}{} }
} if names, err := db.GetPendingEDUServerNames(context.Background()); err == nil {
} else { for _, serverName := range names {
log.WithError(err).Error("Failed to get EDU server names for destination queue hydration") serverNames[serverName] = struct{}{}
} }
for serverName := range serverNames { } else {
if queue := queues.getQueue(serverName); queue != nil { log.WithError(err).Error("Failed to get EDU server names for destination queue hydration")
queue.wakeQueueIfNeeded() }
} offset := time.Duration(5)
for serverName := range serverNames {
if queue := queues.getQueue(serverName); queue != nil {
time.AfterFunc(time.Second*offset, queue.wakeQueueIfNeeded)
offset++
} }
}) }
} }
return queues return queues
} }