MirrorHub/synapse
0
0
Fork 1
mirror of https://mau.dev/maunium/synapse.git synced 2024-12-14 18:43:53 +01:00
Code Issues Projects Releases Wiki Activity

Fix perf of wait_for_stream_positions (#16148)

Browse source
This commit is contained in:
Erik Johnston 2023-08-22 16:11:22 +01:00 committed by GitHub
parent 0ba17777be
commit 803f63df1c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 13 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
changelog.d/16148.bugfix Normal file
View file

@ -0,0 +1 @@
Fix performance degredation when there are a lot of in-flight replication requests.

19
synapse/replication/tcp/client.py
View file

@ -14,7 +14,9 @@
"""A replication client for use by synapse workers. """A replication client for use by synapse workers.
""" """
import logging import logging
from typing import TYPE_CHECKING, Dict, Iterable, List, Optional, Set, Tuple from typing import TYPE_CHECKING, Dict, Iterable, Optional, Set, Tuple
from sortedcontainers import SortedList
from twisted.internet import defer from twisted.internet import defer
from twisted.internet.defer import Deferred from twisted.internet.defer import Deferred
@ -86,7 +88,9 @@ class ReplicationDataHandler:
# Map from stream and instance to list of deferreds waiting for the stream to # Map from stream and instance to list of deferreds waiting for the stream to
# arrive at a particular position. The lists are sorted by stream position. # arrive at a particular position. The lists are sorted by stream position.
self._streams_to_waiters: Dict[Tuple[str, str], List[Tuple[int, Deferred]]] = {} self._streams_to_waiters: Dict[
Tuple[str, str], SortedList[Tuple[int, Deferred]]
] = {}
async def on_rdata( async def on_rdata(
self, stream_name: str, instance_name: str, token: int, rows: list self, stream_name: str, instance_name: str, token: int, rows: list
@ -238,7 +242,9 @@ class ReplicationDataHandler:
# Notify any waiting deferreds. The list is ordered by position so we # Notify any waiting deferreds. The list is ordered by position so we
# just iterate through the list until we reach a position that is # just iterate through the list until we reach a position that is
# greater than the received row position. # greater than the received row position.
waiting_list = self._streams_to_waiters.get((stream_name, instance_name), []) waiting_list = self._streams_to_waiters.get((stream_name, instance_name))
if not waiting_list:
return
# Index of first item with a position after the current token, i.e we # Index of first item with a position after the current token, i.e we
# have called all deferreds before this index. If not overwritten by # have called all deferreds before this index. If not overwritten by
@ -262,7 +268,7 @@ class ReplicationDataHandler:
# Drop all entries in the waiting list that were called in the above # Drop all entries in the waiting list that were called in the above
# loop. (This maintains the order so no need to resort) # loop. (This maintains the order so no need to resort)
waiting_list[:] = waiting_list[index_of_first_deferred_not_called:] del waiting_list[:index_of_first_deferred_not_called]
for deferred in deferreds_to_callback: for deferred in deferreds_to_callback:
try: try:
@ -322,11 +328,10 @@ class ReplicationDataHandler:
) )
waiting_list = self._streams_to_waiters.setdefault( waiting_list = self._streams_to_waiters.setdefault(
(stream_name, instance_name), [] (stream_name, instance_name), SortedList(key=lambda t: t[0])
) )
waiting_list.append((position, deferred)) waiting_list.add((position, deferred))
waiting_list.sort(key=lambda t: t[0])
# We measure here to get in flight counts and average waiting time. # We measure here to get in flight counts and average waiting time.
with Measure(self._clock, "repl.wait_for_stream_position"): with Measure(self._clock, "repl.wait_for_stream_position"):