maxmustermann/synapse
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

more edu batching hackery

Browse source
This commit is contained in:
Richard van der Hoff 2019-02-24 23:15:09 +00:00
parent 4d07dc0d18
commit eb0334b07c
1 changed files with 42 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

54
synapse/federation/transaction_queue.py
View file

@ -66,6 +66,9 @@ sent_edus_by_type = Counter(
["type"],
)
# number of seconds to wait to batch up outgoing EDUs
EDU_BATCH_TIME = 5.0
class TransactionQueue(object):
"""This class makes sure we only have one transaction in flight at
@ -119,6 +122,12 @@ class TransactionQueue(object):
# Map of destination -> (edu_type, key) -> Edu
self.pending_edus_keyed_by_dest = edus_keyed = {}
# In order to batch outgoing EDUs, we delay sending them. This records the time
# when we should send the next batch, by destination.
self.edu_tx_time_by_dest = {}
self.edu_tx_task_by_dest = {}
LaterGauge(
"synapse_federation_transaction_queue_pending_pdus",
"",
@ -380,9 +389,18 @@ class TransactionQueue(object):
else:
self.pending_edus_by_dest.setdefault(destination, []).append(edu)
# this is a bit of a hack, but we delay starting the transmission loop
# in an effort to batch up outgoing EDUs a bit.
self.clock.call_later(5.0, self._attempt_new_transaction, destination)
if destination in self.edu_tx_task_by_dest:
# we already have a job queued to send EDUs to this destination
return
def send_edus():
del self.edu_tx_task_by_dest[destination]
self._send_new_transaction(destination)
self.edu_tx_time_by_dest = self.clock.time() + EDU_BATCH_TIME * 1000
self.edu_tx_task_by_dest[destination] = self.clock.call_later(
EDU_BATCH_TIME, send_edus,
)
def send_device_messages(self, destination):
if destination == self.server_name:
@ -407,6 +425,7 @@ class TransactionQueue(object):
Returns:
None
"""
# list of (pending_pdu, deferred, order)
if destination in self.pending_transactions:
# XXX: pending_transactions can get stuck on by a never-ending
@ -460,19 +479,30 @@ class TransactionQueue(object):
if leftover_pdus:
self.pending_pdus_by_dest[destination] = leftover_pdus
pending_edus = self.pending_edus_by_dest.pop(destination, [])
# if we have PDUs to send, we may as well send EDUs too. Otherwise,
# we only send EDUs if their delay is up
if pending_pdus or (
destination in self.edu_tx_time_by_dest
and self.clock.time() > self.edu_tx_time_by_dest[destination]
):
del self.edu_tx_time_by_dest[destination]
# We can only include at most 100 EDUs per transactions
pending_edus, leftover_edus = pending_edus[:100], pending_edus[100:]
if leftover_edus:
self.pending_edus_by_dest[destination] = leftover_edus
pending_edus = self.pending_edus_by_dest.pop(destination, [])
# We can only include at most 100 EDUs per transactions
pending_edus, leftover_edus = pending_edus[:100], pending_edus[100:]
if leftover_edus:
self.edu_tx_time_by_dest[destination] = self.clock.time()
self.pending_edus_by_dest[destination] = leftover_edus
pending_edus.extend(
self.pending_edus_keyed_by_dest.pop(destination, {}).values()
)
else:
pending_edus = []
pending_presence = self.pending_presence_by_dest.pop(destination, {})
pending_edus.extend(
self.pending_edus_keyed_by_dest.pop(destination, {}).values()
)
pending_edus.extend(device_message_edus)
if pending_presence:
pending_edus.append(