From 79eded1ae4f0d341918f5c58076f1e60cd400e8a Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 21 Sep 2018 14:52:21 +0100
Subject: Make ExpiringCache slightly more performant

---
 synapse/util/caches/expiringcache.py | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

(limited to 'synapse/util')

diff --git a/synapse/util/caches/expiringcache.py b/synapse/util/caches/expiringcache.py
index 921a9c5b29..48ca2d634d 100644
--- a/synapse/util/caches/expiringcache.py
+++ b/synapse/util/caches/expiringcache.py
@@ -16,6 +16,8 @@
 import logging
 from collections import OrderedDict
 
+from six import iteritems
+
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.util.caches import register_cache
 
@@ -127,7 +129,7 @@ class ExpiringCache(object):
 
         keys_to_delete = set()
 
-        for key, cache_entry in self._cache.items():
+        for key, cache_entry in iteritems(self._cache):
             if now - cache_entry.time > self._expiry_ms:
                 keys_to_delete.add(key)
 
@@ -149,6 +151,8 @@ class ExpiringCache(object):
 
 
 class _CacheEntry(object):
+    __slots__ = ["time", "value"]
+
     def __init__(self, time, value):
         self.time = time
         self.value = value
-- 
cgit 1.5.1


From fdd1a62e8d09ddccbe685fe7d7840990a9c06241 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 21 Sep 2018 14:55:47 +0100
Subject: Add a five minute cache to get_destination_retry_timings

Hopefully helps with #3931
---
 synapse/storage/transactions.py      | 23 ++++++++++++++++++++++-
 synapse/util/caches/expiringcache.py | 13 +++++++++++++
 2 files changed, 35 insertions(+), 1 deletion(-)

(limited to 'synapse/util')

diff --git a/synapse/storage/transactions.py b/synapse/storage/transactions.py
index baf0379a68..ab54977a75 100644
--- a/synapse/storage/transactions.py
+++ b/synapse/storage/transactions.py
@@ -23,6 +23,7 @@ from canonicaljson import encode_canonical_json
 from twisted.internet import defer
 
 from synapse.metrics.background_process_metrics import run_as_background_process
+from synapse.util.caches.expiringcache import ExpiringCache
 
 from ._base import SQLBaseStore, db_to_json
 
@@ -49,6 +50,8 @@ _UpdateTransactionRow = namedtuple(
     )
 )
 
+SENTINEL = object()
+
 
 class TransactionStore(SQLBaseStore):
     """A collection of queries for handling PDUs.
@@ -59,6 +62,12 @@ class TransactionStore(SQLBaseStore):
 
         self._clock.looping_call(self._start_cleanup_transactions, 30 * 60 * 1000)
 
+        self._destination_retry_cache = ExpiringCache(
+            cache_name="get_destination_retry_timings",
+            clock=self._clock,
+            expiry_ms=5 * 60 * 1000,
+        )
+
     def get_received_txn_response(self, transaction_id, origin):
         """For an incoming transaction from a given origin, check if we have
         already responded to it. If so, return the response code and response
@@ -155,6 +164,7 @@ class TransactionStore(SQLBaseStore):
         """
         pass
 
+    @defer.inlineCallbacks
     def get_destination_retry_timings(self, destination):
         """Gets the current retry timings (if any) for a given destination.
 
@@ -165,10 +175,20 @@ class TransactionStore(SQLBaseStore):
             None if not retrying
             Otherwise a dict for the retry scheme
         """
-        return self.runInteraction(
+
+        result = self._destination_retry_cache.get(destination, SENTINEL)
+        if result is not SENTINEL:
+            defer.returnValue(result)
+
+        result = yield self.runInteraction(
             "get_destination_retry_timings",
             self._get_destination_retry_timings, destination)
 
+        # We don't hugely care about race conditions between getting and
+        # invalidating the cache, since we time out fairly quickly anyway.
+        self._destination_retry_cache[destination] = result
+        defer.returnValue(result)
+
     def _get_destination_retry_timings(self, txn, destination):
         result = self._simple_select_one_txn(
             txn,
@@ -196,6 +216,7 @@ class TransactionStore(SQLBaseStore):
             retry_interval (int) - how long until next retry in ms
         """
 
+        self._destination_retry_cache.pop(destination)
         return self.runInteraction(
             "set_destination_retry_timings",
             self._set_destination_retry_timings,
diff --git a/synapse/util/caches/expiringcache.py b/synapse/util/caches/expiringcache.py
index 48ca2d634d..346669e4ce 100644
--- a/synapse/util/caches/expiringcache.py
+++ b/synapse/util/caches/expiringcache.py
@@ -24,6 +24,9 @@ from synapse.util.caches import register_cache
 logger = logging.getLogger(__name__)
 
 
+SENTINEL = object()
+
+
 class ExpiringCache(object):
     def __init__(self, cache_name, clock, max_len=0, expiry_ms=0,
                  reset_expiry_on_get=False, iterable=False):
@@ -102,6 +105,16 @@ class ExpiringCache(object):
 
         return entry.value
 
+    def pop(self, key, default=None):
+        value = self._cache.pop(key, SENTINEL)
+        if value is SENTINEL:
+            return default
+
+        if self.iterable:
+            self._size_estimate -= len(value.value)
+
+        return value
+
     def __contains__(self, key):
         return key in self._cache
 
-- 
cgit 1.5.1


From 19dc676d1aaf66a45c9a1810132952f7ae3ca2aa Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 21 Sep 2018 16:25:42 +0100
Subject: Fix ExpiringCache.__len__ to be accurate

It used to try and produce an estimate, which was sometimes negative.
This caused metrics to be sad, so lets always just calculate it from
scratch.
---
 synapse/util/caches/expiringcache.py | 21 +++++++++------------
 1 file changed, 9 insertions(+), 12 deletions(-)

(limited to 'synapse/util')

diff --git a/synapse/util/caches/expiringcache.py b/synapse/util/caches/expiringcache.py
index 346669e4ce..8ac56d85ff 100644
--- a/synapse/util/caches/expiringcache.py
+++ b/synapse/util/caches/expiringcache.py
@@ -16,7 +16,7 @@
 import logging
 from collections import OrderedDict
 
-from six import iteritems
+from six import iteritems, itervalues
 
 from synapse.metrics.background_process_metrics import run_as_background_process
 from synapse.util.caches import register_cache
@@ -59,8 +59,6 @@ class ExpiringCache(object):
 
         self.iterable = iterable
 
-        self._size_estimate = 0
-
         self.metrics = register_cache("expiring", cache_name, self)
 
         if not self._expiry_ms:
@@ -79,16 +77,11 @@ class ExpiringCache(object):
         now = self._clock.time_msec()
         self._cache[key] = _CacheEntry(now, value)
 
-        if self.iterable:
-            self._size_estimate += len(value)
-
         # Evict if there are now too many items
         while self._max_len and len(self) > self._max_len:
             _key, value = self._cache.popitem(last=False)
             if self.iterable:
-                removed_len = len(value.value)
-                self.metrics.inc_evictions(removed_len)
-                self._size_estimate -= removed_len
+                self.metrics.inc_evictions(len(value.value))
             else:
                 self.metrics.inc_evictions()
 
@@ -111,7 +104,9 @@ class ExpiringCache(object):
             return default
 
         if self.iterable:
-            self._size_estimate -= len(value.value)
+            self.metrics.inc_evictions(len(value.value))
+        else:
+            self.metrics.inc_evictions()
 
         return value
 
@@ -149,7 +144,9 @@ class ExpiringCache(object):
         for k in keys_to_delete:
             value = self._cache.pop(k)
             if self.iterable:
-                self._size_estimate -= len(value.value)
+                self.metrics.inc_evictions(len(value.value))
+            else:
+                self.metrics.inc_evictions()
 
         logger.debug(
             "[%s] _prune_cache before: %d, after len: %d",
@@ -158,7 +155,7 @@ class ExpiringCache(object):
 
     def __len__(self):
         if self.iterable:
-            return self._size_estimate
+            return sum(len(entry.value) for entry in itervalues(self._cache))
         else:
             return len(self._cache)
 
-- 
cgit 1.5.1


From 4a15a3e4d539dcea9a4a57e7cd800a926f2a17c3 Mon Sep 17 00:00:00 2001
From: Richard van der Hoff <1389908+richvdh@users.noreply.github.com>
Date: Thu, 27 Sep 2018 11:25:34 +0100
Subject: Include eventid in log lines when processing incoming federation
 transactions (#3959)

when processing incoming transactions, it can be hard to see what's going on,
because we process a bunch of stuff in parallel, and because we may end up
recursively working our way through a chain of three or four events.

This commit creates a way to use logcontexts to add the relevant event ids to
the log lines.
---
 changelog.d/3959.feature                |  1 +
 synapse/federation/federation_server.py | 32 ++++++++--------
 synapse/handlers/federation.py          | 65 ++++++++++++++++++++-------------
 synapse/util/logcontext.py              | 41 +++++++++++++++++++--
 tests/test_federation.py                | 28 ++++++++------
 tests/util/test_logcontext.py           |  5 +++
 6 files changed, 115 insertions(+), 57 deletions(-)
 create mode 100644 changelog.d/3959.feature

(limited to 'synapse/util')

diff --git a/changelog.d/3959.feature b/changelog.d/3959.feature
new file mode 100644
index 0000000000..b3a4f37a8d
--- /dev/null
+++ b/changelog.d/3959.feature
@@ -0,0 +1 @@
+Include eventid in log lines when processing incoming federation transactions
\ No newline at end of file
diff --git a/synapse/federation/federation_server.py b/synapse/federation/federation_server.py
index 9a571e4fc7..819e8f7331 100644
--- a/synapse/federation/federation_server.py
+++ b/synapse/federation/federation_server.py
@@ -46,6 +46,7 @@ from synapse.replication.http.federation import (
 from synapse.types import get_domain_from_id
 from synapse.util.async_helpers import Linearizer, concurrently_execute
 from synapse.util.caches.response_cache import ResponseCache
+from synapse.util.logcontext import nested_logging_context
 from synapse.util.logutils import log_function
 
 # when processing incoming transactions, we try to handle multiple rooms in
@@ -187,21 +188,22 @@ class FederationServer(FederationBase):
 
             for pdu in pdus_by_room[room_id]:
                 event_id = pdu.event_id
-                try:
-                    yield self._handle_received_pdu(
-                        origin, pdu
-                    )
-                    pdu_results[event_id] = {}
-                except FederationError as e:
-                    logger.warn("Error handling PDU %s: %s", event_id, e)
-                    pdu_results[event_id] = {"error": str(e)}
-                except Exception as e:
-                    f = failure.Failure()
-                    pdu_results[event_id] = {"error": str(e)}
-                    logger.error(
-                        "Failed to handle PDU %s: %s",
-                        event_id, f.getTraceback().rstrip(),
-                    )
+                with nested_logging_context(event_id):
+                    try:
+                        yield self._handle_received_pdu(
+                            origin, pdu
+                        )
+                        pdu_results[event_id] = {}
+                    except FederationError as e:
+                        logger.warn("Error handling PDU %s: %s", event_id, e)
+                        pdu_results[event_id] = {"error": str(e)}
+                    except Exception as e:
+                        f = failure.Failure()
+                        pdu_results[event_id] = {"error": str(e)}
+                        logger.error(
+                            "Failed to handle PDU %s: %s",
+                            event_id, f.getTraceback().rstrip(),
+                        )
 
         yield concurrently_execute(
             process_pdus_for_room, pdus_by_room.keys(),
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index 2ccdc3bfa7..993546387c 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -339,14 +339,18 @@ class FederationHandler(BaseHandler):
                             "[%s %s] Requesting state at missing prev_event %s",
                             room_id, event_id, p,
                         )
-                        state, got_auth_chain = (
-                            yield self.federation_client.get_state_for_room(
-                                origin, room_id, p,
+
+                        with logcontext.nested_logging_context(p):
+                            state, got_auth_chain = (
+                                yield self.federation_client.get_state_for_room(
+                                    origin, room_id, p,
+                                )
                             )
-                        )
-                        auth_chains.update(got_auth_chain)
-                        state_group = {(x.type, x.state_key): x.event_id for x in state}
-                        state_groups.append(state_group)
+                            auth_chains.update(got_auth_chain)
+                            state_group = {
+                                (x.type, x.state_key): x.event_id for x in state
+                            }
+                            state_groups.append(state_group)
 
                     # Resolve any conflicting state
                     def fetch(ev_ids):
@@ -483,20 +487,21 @@ class FederationHandler(BaseHandler):
                 "[%s %s] Handling received prev_event %s",
                 room_id, event_id, ev.event_id,
             )
-            try:
-                yield self.on_receive_pdu(
-                    origin,
-                    ev,
-                    sent_to_us_directly=False,
-                )
-            except FederationError as e:
-                if e.code == 403:
-                    logger.warn(
-                        "[%s %s] Received prev_event %s failed history check.",
-                        room_id, event_id, ev.event_id,
+            with logcontext.nested_logging_context(ev.event_id):
+                try:
+                    yield self.on_receive_pdu(
+                        origin,
+                        ev,
+                        sent_to_us_directly=False,
                     )
-                else:
-                    raise
+                except FederationError as e:
+                    if e.code == 403:
+                        logger.warn(
+                            "[%s %s] Received prev_event %s failed history check.",
+                            room_id, event_id, ev.event_id,
+                        )
+                    else:
+                        raise
 
     @defer.inlineCallbacks
     def _process_received_pdu(self, origin, event, state, auth_chain):
@@ -1135,7 +1140,8 @@ class FederationHandler(BaseHandler):
             try:
                 logger.info("Processing queued PDU %s which was received "
                             "while we were joining %s", p.event_id, p.room_id)
-                yield self.on_receive_pdu(origin, p, sent_to_us_directly=True)
+                with logcontext.nested_logging_context(p.event_id):
+                    yield self.on_receive_pdu(origin, p, sent_to_us_directly=True)
             except Exception as e:
                 logger.warn(
                     "Error handling queued PDU %s from %s: %s",
@@ -1581,15 +1587,22 @@ class FederationHandler(BaseHandler):
 
         Notifies about the events where appropriate.
         """
-        contexts = yield logcontext.make_deferred_yieldable(defer.gatherResults(
-            [
-                logcontext.run_in_background(
-                    self._prep_event,
+
+        @defer.inlineCallbacks
+        def prep(ev_info):
+            event = ev_info["event"]
+            with logcontext.nested_logging_context(suffix=event.event_id):
+                res = yield self._prep_event(
                     origin,
-                    ev_info["event"],
+                    event,
                     state=ev_info.get("state"),
                     auth_events=ev_info.get("auth_events"),
                 )
+            defer.returnValue(res)
+
+        contexts = yield logcontext.make_deferred_yieldable(defer.gatherResults(
+            [
+                logcontext.run_in_background(prep, ev_info)
                 for ev_info in event_infos
             ], consumeErrors=True,
         ))
diff --git a/synapse/util/logcontext.py b/synapse/util/logcontext.py
index a0c2d37610..89224b26cc 100644
--- a/synapse/util/logcontext.py
+++ b/synapse/util/logcontext.py
@@ -200,7 +200,7 @@ class LoggingContext(object):
 
     sentinel = Sentinel()
 
-    def __init__(self, name=None, parent_context=None):
+    def __init__(self, name=None, parent_context=None, request=None):
         self.previous_context = LoggingContext.current_context()
         self.name = name
 
@@ -218,6 +218,13 @@ class LoggingContext(object):
 
         self.parent_context = parent_context
 
+        if self.parent_context is not None:
+            self.parent_context.copy_to(self)
+
+        if request is not None:
+            # the request param overrides the request from the parent context
+            self.request = request
+
     def __str__(self):
         return "%s@%x" % (self.name, id(self))
 
@@ -256,9 +263,6 @@ class LoggingContext(object):
             )
         self.alive = True
 
-        if self.parent_context is not None:
-            self.parent_context.copy_to(self)
-
         return self
 
     def __exit__(self, type, value, traceback):
@@ -439,6 +443,35 @@ class PreserveLoggingContext(object):
                 )
 
 
+def nested_logging_context(suffix, parent_context=None):
+    """Creates a new logging context as a child of another.
+
+    The nested logging context will have a 'request' made up of the parent context's
+    request, plus the given suffix.
+
+    CPU/db usage stats will be added to the parent context's on exit.
+
+    Normal usage looks like:
+
+        with nested_logging_context(suffix):
+            # ... do stuff
+
+    Args:
+        suffix (str): suffix to add to the parent context's 'request'.
+        parent_context (LoggingContext|None): parent context. Will use the current context
+            if None.
+
+    Returns:
+        LoggingContext: new logging context.
+    """
+    if parent_context is None:
+        parent_context = LoggingContext.current_context()
+    return LoggingContext(
+        parent_context=parent_context,
+        request=parent_context.request + "-" + suffix,
+    )
+
+
 def preserve_fn(f):
     """Function decorator which wraps the function with run_in_background"""
     def g(*args, **kwargs):
diff --git a/tests/test_federation.py b/tests/test_federation.py
index 2540604fcc..ff55c7a627 100644
--- a/tests/test_federation.py
+++ b/tests/test_federation.py
@@ -6,6 +6,7 @@ from twisted.internet.defer import maybeDeferred, succeed
 from synapse.events import FrozenEvent
 from synapse.types import Requester, UserID
 from synapse.util import Clock
+from synapse.util.logcontext import LoggingContext
 
 from tests import unittest
 from tests.server import ThreadedMemoryReactorClock, setup_test_homeserver
@@ -117,9 +118,10 @@ class MessageAcceptTests(unittest.TestCase):
             }
         )
 
-        d = self.handler.on_receive_pdu(
-            "test.serv", lying_event, sent_to_us_directly=True
-        )
+        with LoggingContext(request="lying_event"):
+            d = self.handler.on_receive_pdu(
+                "test.serv", lying_event, sent_to_us_directly=True
+            )
 
         # Step the reactor, so the database fetches come back
         self.reactor.advance(1)
@@ -209,11 +211,12 @@ class MessageAcceptTests(unittest.TestCase):
             }
         )
 
-        d = self.handler.on_receive_pdu(
-            "test.serv", good_event, sent_to_us_directly=True
-        )
-        self.reactor.advance(1)
-        self.assertEqual(self.successResultOf(d), None)
+        with LoggingContext(request="good_event"):
+            d = self.handler.on_receive_pdu(
+                "test.serv", good_event, sent_to_us_directly=True
+            )
+            self.reactor.advance(1)
+            self.assertEqual(self.successResultOf(d), None)
 
         bad_event = FrozenEvent(
             {
@@ -230,10 +233,11 @@ class MessageAcceptTests(unittest.TestCase):
             }
         )
 
-        d = self.handler.on_receive_pdu(
-            "test.serv", bad_event, sent_to_us_directly=True
-        )
-        self.reactor.advance(1)
+        with LoggingContext(request="bad_event"):
+            d = self.handler.on_receive_pdu(
+                "test.serv", bad_event, sent_to_us_directly=True
+            )
+            self.reactor.advance(1)
 
         extrem = maybeDeferred(
             self.homeserver.datastore.get_latest_event_ids_in_room, self.room_id
diff --git a/tests/util/test_logcontext.py b/tests/util/test_logcontext.py
index 4633db77b3..8adaee3c8d 100644
--- a/tests/util/test_logcontext.py
+++ b/tests/util/test_logcontext.py
@@ -159,6 +159,11 @@ class LoggingContextTestCase(unittest.TestCase):
             self.assertEqual(r, "bum")
             self._check_test_key("one")
 
+    def test_nested_logging_context(self):
+        with LoggingContext(request="foo"):
+            nested_context = logcontext.nested_logging_context(suffix="bar")
+            self.assertEqual(nested_context.request, "foo-bar")
+
 
 # a function which returns a deferred which has been "called", but
 # which had a function which returned another incomplete deferred on
-- 
cgit 1.5.1


From 8ea887856c1bcb97c90d88fe20f7f82ed7f48967 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 1 Oct 2018 11:59:52 +0100
Subject: Don't update eviction metrics on explicit removal

---
 synapse/util/caches/expiringcache.py | 5 -----
 1 file changed, 5 deletions(-)

(limited to 'synapse/util')

diff --git a/synapse/util/caches/expiringcache.py b/synapse/util/caches/expiringcache.py
index 8ac56d85ff..f2f55ba6c9 100644
--- a/synapse/util/caches/expiringcache.py
+++ b/synapse/util/caches/expiringcache.py
@@ -103,11 +103,6 @@ class ExpiringCache(object):
         if value is SENTINEL:
             return default
 
-        if self.iterable:
-            self.metrics.inc_evictions(len(value.value))
-        else:
-            self.metrics.inc_evictions()
-
         return value
 
     def __contains__(self, key):
-- 
cgit 1.5.1


From 4f3e3ac192f2c5ff37198f50af99bc8fbd57beca Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 1 Oct 2018 12:25:27 +0100
Subject: Correctly match 'dict.pop' api

---
 synapse/util/caches/expiringcache.py | 14 +++++++++++---
 1 file changed, 11 insertions(+), 3 deletions(-)

(limited to 'synapse/util')

diff --git a/synapse/util/caches/expiringcache.py b/synapse/util/caches/expiringcache.py
index f2f55ba6c9..f369780277 100644
--- a/synapse/util/caches/expiringcache.py
+++ b/synapse/util/caches/expiringcache.py
@@ -98,10 +98,18 @@ class ExpiringCache(object):
 
         return entry.value
 
-    def pop(self, key, default=None):
-        value = self._cache.pop(key, SENTINEL)
+    def pop(self, key, default=SENTINEL):
+        """Removes and returns the value with the given key from the cache.
+
+        If the key isn't in the cache then `default` will be returned if
+        specified, otherwise `KeyError` will get raised.
+
+        Identical functionality to `dict.pop(..)`.
+        """
+
+        value = self._cache.pop(key, default)
         if value is SENTINEL:
-            return default
+            raise KeyError(key)
 
         return value
 
-- 
cgit 1.5.1


From f7199e873448794ffc88a08a9d7c01f1be0dfca1 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Fri, 5 Oct 2018 11:23:08 +0100
Subject: Log looping call exceptions

If a looping call function errors, then it kills the loop entirely.
Currently it throws away the exception logs, so we should make it
actually log them.

Fixes #3929
---
 synapse/util/__init__.py | 19 ++++++++++++++++++-
 1 file changed, 18 insertions(+), 1 deletion(-)

(limited to 'synapse/util')

diff --git a/synapse/util/__init__.py b/synapse/util/__init__.py
index 680ea928c7..c237d003bc 100644
--- a/synapse/util/__init__.py
+++ b/synapse/util/__init__.py
@@ -13,6 +13,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import functools
 import logging
 from itertools import islice
 
@@ -66,7 +67,7 @@ class Clock(object):
             f(function): The function to call repeatedly.
             msec(float): How long to wait between calls in milliseconds.
         """
-        call = task.LoopingCall(f)
+        call = task.LoopingCall(_log_exception_wrapper(f))
         call.clock = self._reactor
         call.start(msec / 1000.0, now=False)
         return call
@@ -109,3 +110,19 @@ def batch_iter(iterable, size):
     sourceiter = iter(iterable)
     # call islice until it returns an empty tuple
     return iter(lambda: tuple(islice(sourceiter, size)), ())
+
+
+def _log_exception_wrapper(f):
+    """Used to wrap looping calls to log loudly if they get killed
+    """
+
+    @functools.wraps(f)
+    def wrap(*args, **kwargs):
+        try:
+            logger.info("Running looping call")
+            return f(*args, **kwargs)
+        except:  # noqa: E722, as we reraise the exception this is fine.
+            logger.exception("Looping called died")
+            raise
+
+    return wrap
-- 
cgit 1.5.1


From 8a1817f0d29308a233783d43cbf1ad27891120c1 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 Oct 2018 13:26:54 +0100
Subject: Use errback pattern and catch async failures

---
 synapse/handlers/appservice.py |  7 ++++++-
 synapse/util/__init__.py       | 43 ++++++++++++++++++++++++++++--------------
 2 files changed, 35 insertions(+), 15 deletions(-)

(limited to 'synapse/util')

diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py
index f0f89af7dc..16b897eb18 100644
--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -28,6 +28,7 @@ from synapse.metrics import (
     event_processing_loop_room_count,
 )
 from synapse.metrics.background_process_metrics import run_as_background_process
+from synapse.util import make_log_failure_errback
 from synapse.util.logcontext import make_deferred_yieldable, run_in_background
 from synapse.util.metrics import Measure
 
@@ -112,7 +113,11 @@ class ApplicationServicesHandler(object):
 
                         if not self.started_scheduler:
                             def start_scheduler():
-                                return self.scheduler.start().addErrback(log_failure)
+                                return self.scheduler.start().addErrback(
+                                    make_log_failure_errback(
+                                        "Application Services Failure",
+                                    )
+                                )
                             run_as_background_process("as_scheduler", start_scheduler)
                             self.started_scheduler = True
 
diff --git a/synapse/util/__init__.py b/synapse/util/__init__.py
index c237d003bc..964078aed4 100644
--- a/synapse/util/__init__.py
+++ b/synapse/util/__init__.py
@@ -13,7 +13,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import functools
 import logging
 from itertools import islice
 
@@ -67,9 +66,12 @@ class Clock(object):
             f(function): The function to call repeatedly.
             msec(float): How long to wait between calls in milliseconds.
         """
-        call = task.LoopingCall(_log_exception_wrapper(f))
+        call = task.LoopingCall(f)
         call.clock = self._reactor
-        call.start(msec / 1000.0, now=False)
+        d = call.start(msec / 1000.0, now=False)
+        d.addErrback(make_log_failure_errback(
+            "Looping call died", consumeErrors=False,
+        ))
         return call
 
     def call_later(self, delay, callback, *args, **kwargs):
@@ -112,17 +114,30 @@ def batch_iter(iterable, size):
     return iter(lambda: tuple(islice(sourceiter, size)), ())
 
 
-def _log_exception_wrapper(f):
-    """Used to wrap looping calls to log loudly if they get killed
+def make_log_failure_errback(msg, consumeErrors=True):
+    """Creates a function suitable for passing to `Deferred.addErrback` that
+    logs any failures that occur.
+
+    Args:
+        msg (str): Message to log
+        consumeErrors (bool): If true consumes the failure, otherwise passes
+            on down the callback chain
+
+    Returns:
+        func(Failure)
     """
 
-    @functools.wraps(f)
-    def wrap(*args, **kwargs):
-        try:
-            logger.info("Running looping call")
-            return f(*args, **kwargs)
-        except:  # noqa: E722, as we reraise the exception this is fine.
-            logger.exception("Looping called died")
-            raise
+    def log_failure(failure):
+        logger.error(
+            msg,
+            exc_info=(
+                failure.type,
+                failure.value,
+                failure.getTracebackObject()
+            )
+        )
+
+        if not consumeErrors:
+            return failure
 
-    return wrap
+    return log_failure
-- 
cgit 1.5.1


From 69823205722662a72e4203ec304ff595a0f6ecf7 Mon Sep 17 00:00:00 2001
From: Erik Johnston <erik@matrix.org>
Date: Mon, 8 Oct 2018 14:06:19 +0100
Subject: Remove unnecessary extra function call layer

---
 synapse/handlers/appservice.py | 18 +++---------------
 synapse/util/__init__.py       | 29 +++++++++++++----------------
 2 files changed, 16 insertions(+), 31 deletions(-)

(limited to 'synapse/util')

diff --git a/synapse/handlers/appservice.py b/synapse/handlers/appservice.py
index 16b897eb18..17eedf4dbf 100644
--- a/synapse/handlers/appservice.py
+++ b/synapse/handlers/appservice.py
@@ -28,7 +28,7 @@ from synapse.metrics import (
     event_processing_loop_room_count,
 )
 from synapse.metrics.background_process_metrics import run_as_background_process
-from synapse.util import make_log_failure_errback
+from synapse.util import log_failure
 from synapse.util.logcontext import make_deferred_yieldable, run_in_background
 from synapse.util.metrics import Measure
 
@@ -37,17 +37,6 @@ logger = logging.getLogger(__name__)
 events_processed_counter = Counter("synapse_handlers_appservice_events_processed", "")
 
 
-def log_failure(failure):
-    logger.error(
-        "Application Services Failure",
-        exc_info=(
-            failure.type,
-            failure.value,
-            failure.getTracebackObject()
-        )
-    )
-
-
 class ApplicationServicesHandler(object):
 
     def __init__(self, hs):
@@ -114,10 +103,9 @@ class ApplicationServicesHandler(object):
                         if not self.started_scheduler:
                             def start_scheduler():
                                 return self.scheduler.start().addErrback(
-                                    make_log_failure_errback(
-                                        "Application Services Failure",
-                                    )
+                                    log_failure, "Application Services Failure",
                                 )
+
                             run_as_background_process("as_scheduler", start_scheduler)
                             self.started_scheduler = True
 
diff --git a/synapse/util/__init__.py b/synapse/util/__init__.py
index 964078aed4..9a8fae0497 100644
--- a/synapse/util/__init__.py
+++ b/synapse/util/__init__.py
@@ -69,9 +69,9 @@ class Clock(object):
         call = task.LoopingCall(f)
         call.clock = self._reactor
         d = call.start(msec / 1000.0, now=False)
-        d.addErrback(make_log_failure_errback(
-            "Looping call died", consumeErrors=False,
-        ))
+        d.addErrback(
+            log_failure, "Looping call died", consumeErrors=False,
+        )
         return call
 
     def call_later(self, delay, callback, *args, **kwargs):
@@ -114,7 +114,7 @@ def batch_iter(iterable, size):
     return iter(lambda: tuple(islice(sourceiter, size)), ())
 
 
-def make_log_failure_errback(msg, consumeErrors=True):
+def log_failure(failure, msg, consumeErrors=True):
     """Creates a function suitable for passing to `Deferred.addErrback` that
     logs any failures that occur.
 
@@ -127,17 +127,14 @@ def make_log_failure_errback(msg, consumeErrors=True):
         func(Failure)
     """
 
-    def log_failure(failure):
-        logger.error(
-            msg,
-            exc_info=(
-                failure.type,
-                failure.value,
-                failure.getTracebackObject()
-            )
+    logger.error(
+        msg,
+        exc_info=(
+            failure.type,
+            failure.value,
+            failure.getTracebackObject()
         )
+    )
 
-        if not consumeErrors:
-            return failure
-
-    return log_failure
+    if not consumeErrors:
+        return failure
-- 
cgit 1.5.1