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

Rehydrate more opportunistically

Browse source
This commit is contained in:
Neil Alexander 2020-06-30 15:43:15 +01:00
parent b4bbefd523
commit c8ee57ba07
1 changed files with 13 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

20
federationsender/queue/destinationqueue.go
View file

@ -173,12 +173,11 @@ func (oq *destinationQueue) backgroundSend() {
// e.g. in response to EDUs. // e.g. in response to EDUs.
transactionID := gomatrixserverlib.TransactionID("") transactionID := gomatrixserverlib.TransactionID("")
// If we haven't reached the PDU limit yet then rehydrate the // Retrieve the events in the next transaction. Note that this
// PDU queue from the database. We'll retrieve the events in the // does *not* necessarily mean we will fill the buffer - the
// next transaction. Note that this does *not* necessarily mean // important thing here is that we still continue to send
// we will fill the buffer - the important thing here is that we // transactions in order.
// still continue to send transactions in order. hydrate := func() {
if len(oq.pendingPDUs) < maxPDUsPerTransaction {
txid, pdus, err := oq.db.GetNextTransactionPDUs( txid, pdus, err := oq.db.GetNextTransactionPDUs(
context.TODO(), // context context.TODO(), // context
oq.destination, // server name oq.destination, // server name
@ -186,18 +185,25 @@ func (oq *destinationQueue) backgroundSend() {
) )
if err != nil { if err != nil {
log.WithError(err).Errorf("failed to get next transaction PDUs for server %q", oq.destination) log.WithError(err).Errorf("failed to get next transaction PDUs for server %q", oq.destination)
continue return
} }
transactionID = txid transactionID = txid
oq.pendingPDUs = append(oq.pendingPDUs, pdus...) oq.pendingPDUs = append(oq.pendingPDUs, pdus...)
} }
// If we haven't reached the PDU limit yet then rehydrate the
// PDU queue from the database.
if len(oq.pendingPDUs) < maxPDUsPerTransaction {
hydrate()
}
// Wait either for incoming events, or until we hit an // Wait either for incoming events, or until we hit an
// idle timeout. // idle timeout.
if len(oq.pendingPDUs) == 0 { if len(oq.pendingPDUs) == 0 {
select { select {
case <-oq.incomingPDUs: case <-oq.incomingPDUs:
// There are new PDUs waiting in the database. // There are new PDUs waiting in the database.
hydrate()
case edu := <-oq.incomingEDUs: case edu := <-oq.incomingEDUs:
// Likewise for EDUs, although we should probably not try // Likewise for EDUs, although we should probably not try
// too hard with some EDUs (like typing notifications) after // too hard with some EDUs (like typing notifications) after