From cc50b1ae5343f70547f1000472f9f5355f43a230 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 30 Dec 2016 18:13:15 +0000 Subject: [PATCH 1/4] Remove fallback from get_missing_events. get_missing_events used to fallback to fetching the missing events individually requesting from every server in the room, one by one.e This could be unacceptably slow, possibly causing #1732 --- synapse/federation/federation_client.py | 69 ------------------------- 1 file changed, 69 deletions(-) diff --git a/synapse/federation/federation_client.py b/synapse/federation/federation_client.py index 6e23c207e..6851f2376 100644 --- a/synapse/federation/federation_client.py +++ b/synapse/federation/federation_client.py @@ -27,7 +27,6 @@ from synapse.util.caches.expiringcache import ExpiringCache from synapse.util.logutils import log_function from synapse.util.logcontext import preserve_fn, preserve_context_over_deferred from synapse.events import FrozenEvent -from synapse.types import get_domain_from_id import synapse.metrics from synapse.util.retryutils import get_retry_limiter, NotRetryingDestination @@ -741,8 +740,6 @@ class FederationClient(FederationBase): signed_events = yield self._check_sigs_and_hash_and_fetch( destination, events, outlier=False ) - - have_gotten_all_from_destination = True except HttpResponseException as e: if not e.code == 400: raise @@ -750,72 +747,6 @@ class FederationClient(FederationBase): # We are probably hitting an old server that doesn't support # get_missing_events signed_events = [] - have_gotten_all_from_destination = False - - if len(signed_events) >= limit: - defer.returnValue(signed_events) - - users = yield self.state.get_current_user_in_room(room_id) - servers = set(get_domain_from_id(u) for u in users) - - servers = set(servers) - servers.discard(self.server_name) - - failed_to_fetch = set() - - while len(signed_events) < limit: - # Are we missing any? - - seen_events = set(earliest_events_ids) - seen_events.update(e.event_id for e in signed_events if e) - - missing_events = {} - for e in itertools.chain(latest_events, signed_events): - if e.depth > min_depth: - missing_events.update({ - e_id: e.depth for e_id, _ in e.prev_events - if e_id not in seen_events - and e_id not in failed_to_fetch - }) - - if not missing_events: - break - - have_seen = yield self.store.have_events(missing_events) - - for k in have_seen: - missing_events.pop(k, None) - - if not missing_events: - break - - # Okay, we haven't gotten everything yet. Lets get them. - ordered_missing = sorted(missing_events.items(), key=lambda x: x[0]) - - if have_gotten_all_from_destination: - servers.discard(destination) - - def random_server_list(): - srvs = list(servers) - random.shuffle(srvs) - return srvs - - deferreds = [ - preserve_fn(self.get_pdu)( - destinations=random_server_list(), - event_id=e_id, - ) - for e_id, depth in ordered_missing[:limit - len(signed_events)] - ] - - res = yield preserve_context_over_deferred( - defer.DeferredList(deferreds, consumeErrors=True) - ) - for (result, val), (e_id, _) in zip(res, ordered_missing): - if result and val: - signed_events.append(val) - else: - failed_to_fetch.add(e_id) defer.returnValue(signed_events) From 0aff09f6c93193c2f3000e1e70e713f615445fe9 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 30 Dec 2016 20:00:44 +0000 Subject: [PATCH 2/4] Add more useful logging when we block fetching events --- synapse/federation/federation_server.py | 9 +++++++++ synapse/util/async.py | 5 +++++ 2 files changed, 14 insertions(+) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 3fa7b2315..6e600fcb8 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -538,7 +538,16 @@ class FederationServer(FederationBase): if get_missing and prevs - seen: # If we're missing stuff, ensure we only fetch stuff one # at a time. + logger.info( + "Aquiring lock for room %r to fetch %d missing events: %r...", + pdu.room_id, len(prevs - seen), list(prevs - seen)[:5], + ) with (yield self._room_pdu_linearizer.queue(pdu.room_id)): + logger.info( + "Aquired lock for room %r to fetch %d missing events", + pdu.room_id, len(prevs - seen), + ) + # We recalculate seen, since it may have changed. have_seen = yield self.store.have_events(prevs) seen = set(have_seen.keys()) diff --git a/synapse/util/async.py b/synapse/util/async.py index 16ed183d4..1e124163f 100644 --- a/synapse/util/async.py +++ b/synapse/util/async.py @@ -23,6 +23,10 @@ from synapse.util import unwrapFirstError from contextlib import contextmanager +import logging + +logger = logging.getLogger(__name__) + @defer.inlineCallbacks def sleep(seconds): @@ -181,6 +185,7 @@ class Linearizer(object): self.key_to_defer[key] = new_defer if current_defer: + logger.info("Waiting to aquire linearizer lock for key %r", key) with PreserveLoggingContext(): yield current_defer From 62ce3034f3096d7a7e920200d2a7cbd5534f655b Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 30 Dec 2016 20:04:44 +0000 Subject: [PATCH 3/4] s/aquire/acquire/g --- synapse/federation/federation_server.py | 4 ++-- synapse/util/async.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py index 6e600fcb8..f4c60e67e 100644 --- a/synapse/federation/federation_server.py +++ b/synapse/federation/federation_server.py @@ -539,12 +539,12 @@ class FederationServer(FederationBase): # If we're missing stuff, ensure we only fetch stuff one # at a time. logger.info( - "Aquiring lock for room %r to fetch %d missing events: %r...", + "Acquiring lock for room %r to fetch %d missing events: %r...", pdu.room_id, len(prevs - seen), list(prevs - seen)[:5], ) with (yield self._room_pdu_linearizer.queue(pdu.room_id)): logger.info( - "Aquired lock for room %r to fetch %d missing events", + "Acquired lock for room %r to fetch %d missing events", pdu.room_id, len(prevs - seen), ) diff --git a/synapse/util/async.py b/synapse/util/async.py index 1e124163f..4280455cb 100644 --- a/synapse/util/async.py +++ b/synapse/util/async.py @@ -185,7 +185,7 @@ class Linearizer(object): self.key_to_defer[key] = new_defer if current_defer: - logger.info("Waiting to aquire linearizer lock for key %r", key) + logger.info("Waiting to acquire linearizer lock for key %r", key) with PreserveLoggingContext(): yield current_defer From f3de4f8cb7383a71bbbdd4c3c45dd803ee45e415 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 30 Dec 2016 20:21:04 +0000 Subject: [PATCH 4/4] Bump version and changelog --- CHANGES.rst | 10 ++++++++++ synapse/__init__.py | 2 +- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/CHANGES.rst b/CHANGES.rst index 108f827cf..dfa5d8748 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -1,3 +1,13 @@ +Changes in synapse v0.18.6-rc2 (2016-12-30) +=========================================== + +Bug fixes: + +* Fix memory leak in twisted by initialising logging correctly (PR #1731) +* Fix bug where fetching missing events took an unacceptable amount of time in + large rooms (PR #1734) + + Changes in synapse v0.18.6-rc1 (2016-12-29) =========================================== diff --git a/synapse/__init__.py b/synapse/__init__.py index 84592f53e..fbb2bb57b 100644 --- a/synapse/__init__.py +++ b/synapse/__init__.py @@ -16,4 +16,4 @@ """ This is a reference implementation of a Matrix home server. """ -__version__ = "0.18.6-rc1" +__version__ = "0.18.6-rc2"