diff options
author | Erik Johnston <erik@matrix.org> | 2017-10-23 13:13:53 +0100 |
---|---|---|
committer | Erik Johnston <erik@matrix.org> | 2017-10-23 13:13:53 +0100 |
commit | ffba97807746f6f0d7a8820c0d980f188dcad08c (patch) | |
tree | 36301e98cdc6400afe71b89fc4b0c8beccd0deea /synapse/util/async.py | |
parent | Merge pull request #2540 from 4nd3r/patch-1 (diff) | |
parent | Bump version and changelog (diff) | |
download | synapse-ffba97807746f6f0d7a8820c0d980f188dcad08c.tar.xz |
Merge branch 'release-v0.24.0' of github.com:matrix-org/synapse v0.24.0
Diffstat (limited to 'synapse/util/async.py')
-rw-r--r-- | synapse/util/async.py | 33 |
1 files changed, 29 insertions, 4 deletions
diff --git a/synapse/util/async.py b/synapse/util/async.py index 1453faf0ef..a0a9039475 100644 --- a/synapse/util/async.py +++ b/synapse/util/async.py @@ -19,7 +19,7 @@ from twisted.internet import defer, reactor from .logcontext import ( PreserveLoggingContext, preserve_fn, preserve_context_over_deferred, ) -from synapse.util import unwrapFirstError +from synapse.util import logcontext, unwrapFirstError from contextlib import contextmanager @@ -53,6 +53,11 @@ class ObservableDeferred(object): Cancelling or otherwise resolving an observer will not affect the original ObservableDeferred. + + NB that it does not attempt to do anything with logcontexts; in general + you should probably make_deferred_yieldable the deferreds + returned by `observe`, and ensure that the original deferred runs its + callbacks in the sentinel logcontext. """ __slots__ = ["_deferred", "_observers", "_result"] @@ -155,7 +160,7 @@ def concurrently_execute(func, args, limit): except StopIteration: pass - return preserve_context_over_deferred(defer.gatherResults([ + return logcontext.make_deferred_yieldable(defer.gatherResults([ preserve_fn(_concurrently_execute_inner)() for _ in xrange(limit) ], consumeErrors=True)).addErrback(unwrapFirstError) @@ -203,7 +208,26 @@ class Linearizer(object): except: logger.exception("Unexpected exception in Linearizer") - logger.info("Acquired linearizer lock %r for key %r", self.name, key) + logger.info("Acquired linearizer lock %r for key %r", self.name, + key) + + # if the code holding the lock completes synchronously, then it + # will recursively run the next claimant on the list. That can + # relatively rapidly lead to stack exhaustion. This is essentially + # the same problem as http://twistedmatrix.com/trac/ticket/9304. + # + # In order to break the cycle, we add a cheeky sleep(0) here to + # ensure that we fall back to the reactor between each iteration. + # + # (There's no particular need for it to happen before we return + # the context manager, but it needs to happen while we hold the + # lock, and the context manager's exit code must be synchronous, + # so actually this is the only sensible place. + yield run_on_reactor() + + else: + logger.info("Acquired uncontended linearizer lock %r for key %r", + self.name, key) @contextmanager def _ctx_manager(): @@ -211,7 +235,8 @@ class Linearizer(object): yield finally: logger.info("Releasing linearizer lock %r for key %r", self.name, key) - new_defer.callback(None) + with PreserveLoggingContext(): + new_defer.callback(None) current_d = self.key_to_defer.get(key) if current_d is new_defer: self.key_to_defer.pop(key, None) |