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

Dedupe destinations, fix other bug hopefully

Browse source
This commit is contained in:
Neil Alexander 2020-05-06 14:00:04 +01:00
parent 9d293caec8
commit 18cb75b0ac
2 changed files with 21 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
federationsender/queue/destinationqueue.go
View file

@ -122,7 +122,8 @@ func (oq *destinationQueue) backgroundSend() {
// Retrieve any waiting things. // Retrieve any waiting things.
oq.runningMutex.RLock() oq.runningMutex.RLock()
pendingPDUs, pendingEDUs := oq.pendingPDUs, oq.pendingEDUs pendingPDUs, numPDUs := oq.pendingPDUs, len(oq.pendingPDUs)
pendingEDUs, numEDUs := oq.pendingEDUs, len(oq.pendingEDUs)
pendingInvites := oq.pendingInvites pendingInvites := oq.pendingInvites
idleCounter, sentCounter := oq.idleCounter.Load(), oq.statistics.SuccessCount() idleCounter, sentCounter := oq.idleCounter.Load(), oq.statistics.SuccessCount()
oq.runningMutex.RUnlock() oq.runningMutex.RUnlock()
@ -156,13 +157,19 @@ func (oq *destinationQueue) backgroundSend() {
oq.runningMutex.Lock() oq.runningMutex.Lock()
// Reallocate so that the underlying arrays can be GC'd, as // Reallocate so that the underlying arrays can be GC'd, as
// opposed to growing forever. // opposed to growing forever.
for i := 0; i < numPDUs; i++ {
oq.pendingPDUs[i] = nil
}
for i := 0; i < numEDUs; i++ {
oq.pendingEDUs[i] = nil
}
oq.pendingPDUs = append( oq.pendingPDUs = append(
[]*gomatrixserverlib.HeaderedEvent{}, []*gomatrixserverlib.HeaderedEvent{},
oq.pendingPDUs[len(pendingPDUs):]..., oq.pendingPDUs[numPDUs:]...,
) )
oq.pendingEDUs = append( oq.pendingEDUs = append(
[]*gomatrixserverlib.EDU{}, []*gomatrixserverlib.EDU{},
oq.pendingEDUs[len(pendingEDUs):]..., oq.pendingEDUs[numEDUs:]...,
) )
oq.runningMutex.Unlock() oq.runningMutex.Unlock()
} }
@ -179,7 +186,6 @@ func (oq *destinationQueue) backgroundSend() {
// the backoff has exceeded a maximum allowable value. // the backoff has exceeded a maximum allowable value.
return return
} }
continue
} else if sent > 0 { } else if sent > 0 {
// If we successfully sent the invites then clear out // If we successfully sent the invites then clear out
// the pending invites. // the pending invites.

16
federationsender/queue/queue.go
View file

@ -21,6 +21,7 @@ import (
"github.com/matrix-org/dendrite/federationsender/producers" "github.com/matrix-org/dendrite/federationsender/producers"
"github.com/matrix-org/dendrite/federationsender/types" "github.com/matrix-org/dendrite/federationsender/types"
"github.com/matrix-org/gomatrixserverlib" "github.com/matrix-org/gomatrixserverlib"
"github.com/matrix-org/util"
log "github.com/sirupsen/logrus" log "github.com/sirupsen/logrus"
) )
@ -189,13 +190,18 @@ func (oqs *OutgoingQueues) SendEDU(
// filterDestinations removes our own server from the list of destinations. // filterDestinations removes our own server from the list of destinations.
// Otherwise we could end up trying to talk to ourselves. // Otherwise we could end up trying to talk to ourselves.
func filterDestinations(origin gomatrixserverlib.ServerName, destinations []gomatrixserverlib.ServerName) []gomatrixserverlib.ServerName { func filterDestinations(origin gomatrixserverlib.ServerName, destinations []gomatrixserverlib.ServerName) (
var result []gomatrixserverlib.ServerName result []gomatrixserverlib.ServerName,
for _, destination := range destinations { ) {
if destination == origin { strs := make([]string, len(destinations))
for i, d := range destinations {
strs[i] = string(d)
}
for _, destination := range util.UniqueStrings(strs) {
if gomatrixserverlib.ServerName(destination) == origin {
continue continue
} }
result = append(result, destination) result = append(result, gomatrixserverlib.ServerName(destination))
} }
return result return result
} }