diff options
author | Richard van der Hoff <1389908+richvdh@users.noreply.github.com> | 2018-12-04 10:53:49 +0100 |
---|---|---|
committer | GitHub <noreply@github.com> | 2018-12-04 10:53:49 +0100 |
commit | 52e87fbfbe55e647451d886f02f49d3bb5517274 (patch) | |
tree | 46a6e82662d07e37d36ba7670c23e5bd7ee4a372 /synapse/appservice | |
parent | Workaround for non-ascii event ids (#4241) (diff) | |
download | synapse-52e87fbfbe55e647451d886f02f49d3bb5517274.tar.xz |
Run the AS senders as background processes (#4189)
This should fix some "Starting db connection from sentinel context" warnings, and will mean we get metrics for these processes.
Diffstat (limited to '')
-rw-r--r-- | synapse/appservice/scheduler.py | 36 |
1 files changed, 24 insertions, 12 deletions
diff --git a/synapse/appservice/scheduler.py b/synapse/appservice/scheduler.py index 2430814796..685f15c061 100644 --- a/synapse/appservice/scheduler.py +++ b/synapse/appservice/scheduler.py @@ -53,8 +53,8 @@ import logging from twisted.internet import defer from synapse.appservice import ApplicationServiceState +from synapse.metrics.background_process_metrics import run_as_background_process from synapse.util.logcontext import run_in_background -from synapse.util.metrics import Measure logger = logging.getLogger(__name__) @@ -104,27 +104,34 @@ class _ServiceQueuer(object): self.clock = clock def enqueue(self, service, event): - # if this service isn't being sent something self.queued_events.setdefault(service.id, []).append(event) - run_in_background(self._send_request, service) - @defer.inlineCallbacks - def _send_request(self, service): + # start a sender for this appservice if we don't already have one + if service.id in self.requests_in_flight: return + run_as_background_process( + "as-sender-%s" % (service.id, ), + self._send_request, service, + ) + + @defer.inlineCallbacks + def _send_request(self, service): + # sanity-check: we shouldn't get here if this service already has a sender + # running. + assert(service.id not in self.requests_in_flight) + self.requests_in_flight.add(service.id) try: while True: events = self.queued_events.pop(service.id, []) if not events: return - - with Measure(self.clock, "servicequeuer.send"): - try: - yield self.txn_ctrl.send(service, events) - except Exception: - logger.exception("AS request failed") + try: + yield self.txn_ctrl.send(service, events) + except Exception: + logger.exception("AS request failed") finally: self.requests_in_flight.discard(service.id) @@ -223,7 +230,12 @@ class _Recoverer(object): self.backoff_counter = 1 def recover(self): - self.clock.call_later((2 ** self.backoff_counter), self.retry) + def _retry(): + run_as_background_process( + "as-recoverer-%s" % (self.service.id,), + self.retry, + ) + self.clock.call_later((2 ** self.backoff_counter), _retry) def _backoff(self): # cap the backoff to be around 8.5min => (2^9) = 512 secs |