mirror of
https://github.com/element-hq/synapse
synced 2024-08-19 22:00:22 +00:00
Merge pull request #2675 from matrix-org/rav/remove_broken_logcontext_funcs
Remove preserve_context_over_{fn, deferred}
This commit is contained in:
commit
1b83c09c03
10 changed files with 24 additions and 97 deletions
|
@ -298,10 +298,6 @@ It can be used like this:
|
||||||
# this will now be logged against the request context
|
# this will now be logged against the request context
|
||||||
logger.debug("Request handling complete")
|
logger.debug("Request handling complete")
|
||||||
|
|
||||||
XXX: I think ``preserve_context_over_fn`` is supposed to do the first option,
|
|
||||||
but the fact that it does ``preserve_context_over_deferred`` on its results
|
|
||||||
means that its use is fraught with difficulty.
|
|
||||||
|
|
||||||
Passing synapse deferreds into third-party functions
|
Passing synapse deferreds into third-party functions
|
||||||
----------------------------------------------------
|
----------------------------------------------------
|
||||||
|
|
||||||
|
|
|
@ -25,7 +25,7 @@ from synapse.api.errors import (
|
||||||
from synapse.util import unwrapFirstError, logcontext
|
from synapse.util import unwrapFirstError, logcontext
|
||||||
from synapse.util.caches.expiringcache import ExpiringCache
|
from synapse.util.caches.expiringcache import ExpiringCache
|
||||||
from synapse.util.logutils import log_function
|
from synapse.util.logutils import log_function
|
||||||
from synapse.util.logcontext import preserve_fn, preserve_context_over_deferred
|
from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
|
||||||
from synapse.events import FrozenEvent, builder
|
from synapse.events import FrozenEvent, builder
|
||||||
import synapse.metrics
|
import synapse.metrics
|
||||||
|
|
||||||
|
@ -420,7 +420,7 @@ class FederationClient(FederationBase):
|
||||||
for e_id in batch
|
for e_id in batch
|
||||||
]
|
]
|
||||||
|
|
||||||
res = yield preserve_context_over_deferred(
|
res = yield make_deferred_yieldable(
|
||||||
defer.DeferredList(deferreds, consumeErrors=True)
|
defer.DeferredList(deferreds, consumeErrors=True)
|
||||||
)
|
)
|
||||||
for success, result in res:
|
for success, result in res:
|
||||||
|
|
|
@ -17,7 +17,7 @@ from twisted.internet import defer
|
||||||
|
|
||||||
from synapse.api.constants import EventTypes
|
from synapse.api.constants import EventTypes
|
||||||
from synapse.util.metrics import Measure
|
from synapse.util.metrics import Measure
|
||||||
from synapse.util.logcontext import preserve_fn, preserve_context_over_deferred
|
from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
@ -159,7 +159,7 @@ class ApplicationServicesHandler(object):
|
||||||
def query_3pe(self, kind, protocol, fields):
|
def query_3pe(self, kind, protocol, fields):
|
||||||
services = yield self._get_services_for_3pn(protocol)
|
services = yield self._get_services_for_3pn(protocol)
|
||||||
|
|
||||||
results = yield preserve_context_over_deferred(defer.DeferredList([
|
results = yield make_deferred_yieldable(defer.DeferredList([
|
||||||
preserve_fn(self.appservice_api.query_3pe)(service, kind, protocol, fields)
|
preserve_fn(self.appservice_api.query_3pe)(service, kind, protocol, fields)
|
||||||
for service in services
|
for service in services
|
||||||
], consumeErrors=True))
|
], consumeErrors=True))
|
||||||
|
|
|
@ -27,7 +27,7 @@ from synapse.types import (
|
||||||
from synapse.util import unwrapFirstError
|
from synapse.util import unwrapFirstError
|
||||||
from synapse.util.async import concurrently_execute
|
from synapse.util.async import concurrently_execute
|
||||||
from synapse.util.caches.snapshot_cache import SnapshotCache
|
from synapse.util.caches.snapshot_cache import SnapshotCache
|
||||||
from synapse.util.logcontext import preserve_fn, preserve_context_over_deferred
|
from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
|
||||||
from synapse.visibility import filter_events_for_client
|
from synapse.visibility import filter_events_for_client
|
||||||
|
|
||||||
from ._base import BaseHandler
|
from ._base import BaseHandler
|
||||||
|
@ -163,7 +163,7 @@ class InitialSyncHandler(BaseHandler):
|
||||||
lambda states: states[event.event_id]
|
lambda states: states[event.event_id]
|
||||||
)
|
)
|
||||||
|
|
||||||
(messages, token), current_state = yield preserve_context_over_deferred(
|
(messages, token), current_state = yield make_deferred_yieldable(
|
||||||
defer.gatherResults(
|
defer.gatherResults(
|
||||||
[
|
[
|
||||||
preserve_fn(self.store.get_recent_events_for_room)(
|
preserve_fn(self.store.get_recent_events_for_room)(
|
||||||
|
|
|
@ -17,7 +17,7 @@
|
||||||
from twisted.internet import defer
|
from twisted.internet import defer
|
||||||
|
|
||||||
from .pusher import PusherFactory
|
from .pusher import PusherFactory
|
||||||
from synapse.util.logcontext import preserve_fn, preserve_context_over_deferred
|
from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
|
||||||
from synapse.util.async import run_on_reactor
|
from synapse.util.async import run_on_reactor
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
@ -136,7 +136,7 @@ class PusherPool:
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
yield preserve_context_over_deferred(defer.gatherResults(deferreds))
|
yield make_deferred_yieldable(defer.gatherResults(deferreds))
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("Exception in pusher on_new_notifications")
|
logger.exception("Exception in pusher on_new_notifications")
|
||||||
|
|
||||||
|
@ -161,7 +161,7 @@ class PusherPool:
|
||||||
preserve_fn(p.on_new_receipts)(min_stream_id, max_stream_id)
|
preserve_fn(p.on_new_receipts)(min_stream_id, max_stream_id)
|
||||||
)
|
)
|
||||||
|
|
||||||
yield preserve_context_over_deferred(defer.gatherResults(deferreds))
|
yield make_deferred_yieldable(defer.gatherResults(deferreds))
|
||||||
except Exception:
|
except Exception:
|
||||||
logger.exception("Exception in pusher on_new_receipts")
|
logger.exception("Exception in pusher on_new_receipts")
|
||||||
|
|
||||||
|
|
|
@ -39,7 +39,7 @@ from ._base import SQLBaseStore
|
||||||
from synapse.util.caches.descriptors import cached
|
from synapse.util.caches.descriptors import cached
|
||||||
from synapse.api.constants import EventTypes
|
from synapse.api.constants import EventTypes
|
||||||
from synapse.types import RoomStreamToken
|
from synapse.types import RoomStreamToken
|
||||||
from synapse.util.logcontext import preserve_fn, preserve_context_over_deferred
|
from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
|
||||||
from synapse.storage.engines import PostgresEngine, Sqlite3Engine
|
from synapse.storage.engines import PostgresEngine, Sqlite3Engine
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
@ -234,7 +234,7 @@ class StreamStore(SQLBaseStore):
|
||||||
results = {}
|
results = {}
|
||||||
room_ids = list(room_ids)
|
room_ids = list(room_ids)
|
||||||
for rm_ids in (room_ids[i:i + 20] for i in xrange(0, len(room_ids), 20)):
|
for rm_ids in (room_ids[i:i + 20] for i in xrange(0, len(room_ids), 20)):
|
||||||
res = yield preserve_context_over_deferred(defer.gatherResults([
|
res = yield make_deferred_yieldable(defer.gatherResults([
|
||||||
preserve_fn(self.get_room_events_stream_for_room)(
|
preserve_fn(self.get_room_events_stream_for_room)(
|
||||||
room_id, from_key, to_key, limit, order=order,
|
room_id, from_key, to_key, limit, order=order,
|
||||||
)
|
)
|
||||||
|
|
|
@ -17,7 +17,7 @@
|
||||||
from twisted.internet import defer, reactor
|
from twisted.internet import defer, reactor
|
||||||
|
|
||||||
from .logcontext import (
|
from .logcontext import (
|
||||||
PreserveLoggingContext, preserve_fn, preserve_context_over_deferred,
|
PreserveLoggingContext, make_deferred_yieldable, preserve_fn
|
||||||
)
|
)
|
||||||
from synapse.util import logcontext, unwrapFirstError
|
from synapse.util import logcontext, unwrapFirstError
|
||||||
|
|
||||||
|
@ -351,7 +351,7 @@ class ReadWriteLock(object):
|
||||||
|
|
||||||
# We wait for the latest writer to finish writing. We can safely ignore
|
# We wait for the latest writer to finish writing. We can safely ignore
|
||||||
# any existing readers... as they're readers.
|
# any existing readers... as they're readers.
|
||||||
yield curr_writer
|
yield make_deferred_yieldable(curr_writer)
|
||||||
|
|
||||||
@contextmanager
|
@contextmanager
|
||||||
def _ctx_manager():
|
def _ctx_manager():
|
||||||
|
@ -380,7 +380,7 @@ class ReadWriteLock(object):
|
||||||
curr_readers.clear()
|
curr_readers.clear()
|
||||||
self.key_to_current_writer[key] = new_defer
|
self.key_to_current_writer[key] = new_defer
|
||||||
|
|
||||||
yield preserve_context_over_deferred(defer.gatherResults(to_wait_on))
|
yield make_deferred_yieldable(defer.gatherResults(to_wait_on))
|
||||||
|
|
||||||
@contextmanager
|
@contextmanager
|
||||||
def _ctx_manager():
|
def _ctx_manager():
|
||||||
|
|
|
@ -13,32 +13,24 @@
|
||||||
# See the License for the specific language governing permissions and
|
# See the License for the specific language governing permissions and
|
||||||
# limitations under the License.
|
# limitations under the License.
|
||||||
|
|
||||||
from twisted.internet import defer
|
|
||||||
|
|
||||||
from synapse.util.logcontext import (
|
|
||||||
PreserveLoggingContext, preserve_context_over_fn
|
|
||||||
)
|
|
||||||
|
|
||||||
from synapse.util import unwrapFirstError
|
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
from twisted.internet import defer
|
||||||
|
|
||||||
|
from synapse.util import unwrapFirstError
|
||||||
|
from synapse.util.logcontext import PreserveLoggingContext
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def user_left_room(distributor, user, room_id):
|
def user_left_room(distributor, user, room_id):
|
||||||
return preserve_context_over_fn(
|
with PreserveLoggingContext():
|
||||||
distributor.fire,
|
distributor.fire("user_left_room", user=user, room_id=room_id)
|
||||||
"user_left_room", user=user, room_id=room_id
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def user_joined_room(distributor, user, room_id):
|
def user_joined_room(distributor, user, room_id):
|
||||||
return preserve_context_over_fn(
|
with PreserveLoggingContext():
|
||||||
distributor.fire,
|
distributor.fire("user_joined_room", user=user, room_id=room_id)
|
||||||
"user_joined_room", user=user, room_id=room_id
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class Distributor(object):
|
class Distributor(object):
|
||||||
|
|
|
@ -261,67 +261,6 @@ class PreserveLoggingContext(object):
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
class _PreservingContextDeferred(defer.Deferred):
|
|
||||||
"""A deferred that ensures that all callbacks and errbacks are called with
|
|
||||||
the given logging context.
|
|
||||||
"""
|
|
||||||
def __init__(self, context):
|
|
||||||
self._log_context = context
|
|
||||||
defer.Deferred.__init__(self)
|
|
||||||
|
|
||||||
def addCallbacks(self, callback, errback=None,
|
|
||||||
callbackArgs=None, callbackKeywords=None,
|
|
||||||
errbackArgs=None, errbackKeywords=None):
|
|
||||||
callback = self._wrap_callback(callback)
|
|
||||||
errback = self._wrap_callback(errback)
|
|
||||||
return defer.Deferred.addCallbacks(
|
|
||||||
self, callback,
|
|
||||||
errback=errback,
|
|
||||||
callbackArgs=callbackArgs,
|
|
||||||
callbackKeywords=callbackKeywords,
|
|
||||||
errbackArgs=errbackArgs,
|
|
||||||
errbackKeywords=errbackKeywords,
|
|
||||||
)
|
|
||||||
|
|
||||||
def _wrap_callback(self, f):
|
|
||||||
def g(res, *args, **kwargs):
|
|
||||||
with PreserveLoggingContext(self._log_context):
|
|
||||||
res = f(res, *args, **kwargs)
|
|
||||||
return res
|
|
||||||
return g
|
|
||||||
|
|
||||||
|
|
||||||
def preserve_context_over_fn(fn, *args, **kwargs):
|
|
||||||
"""Takes a function and invokes it with the given arguments, but removes
|
|
||||||
and restores the current logging context while doing so.
|
|
||||||
|
|
||||||
If the result is a deferred, call preserve_context_over_deferred before
|
|
||||||
returning it.
|
|
||||||
"""
|
|
||||||
with PreserveLoggingContext():
|
|
||||||
res = fn(*args, **kwargs)
|
|
||||||
|
|
||||||
if isinstance(res, defer.Deferred):
|
|
||||||
return preserve_context_over_deferred(res)
|
|
||||||
else:
|
|
||||||
return res
|
|
||||||
|
|
||||||
|
|
||||||
def preserve_context_over_deferred(deferred, context=None):
|
|
||||||
"""Given a deferred wrap it such that any callbacks added later to it will
|
|
||||||
be invoked with the current context.
|
|
||||||
|
|
||||||
Deprecated: this almost certainly doesn't do want you want, ie make
|
|
||||||
the deferred follow the synapse logcontext rules: try
|
|
||||||
``make_deferred_yieldable`` instead.
|
|
||||||
"""
|
|
||||||
if context is None:
|
|
||||||
context = LoggingContext.current_context()
|
|
||||||
d = _PreservingContextDeferred(context)
|
|
||||||
deferred.chainDeferred(d)
|
|
||||||
return d
|
|
||||||
|
|
||||||
|
|
||||||
def preserve_fn(f):
|
def preserve_fn(f):
|
||||||
"""Wraps a function, to ensure that the current context is restored after
|
"""Wraps a function, to ensure that the current context is restored after
|
||||||
return from the function, and that the sentinel context is set once the
|
return from the function, and that the sentinel context is set once the
|
||||||
|
|
|
@ -17,7 +17,7 @@ from twisted.internet import defer
|
||||||
|
|
||||||
from synapse.api.constants import Membership, EventTypes
|
from synapse.api.constants import Membership, EventTypes
|
||||||
|
|
||||||
from synapse.util.logcontext import preserve_fn, preserve_context_over_deferred
|
from synapse.util.logcontext import make_deferred_yieldable, preserve_fn
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
@ -58,7 +58,7 @@ def filter_events_for_clients(store, user_tuples, events, event_id_to_state,
|
||||||
always_include_ids (set(event_id)): set of event ids to specifically
|
always_include_ids (set(event_id)): set of event ids to specifically
|
||||||
include (unless sender is ignored)
|
include (unless sender is ignored)
|
||||||
"""
|
"""
|
||||||
forgotten = yield preserve_context_over_deferred(defer.gatherResults([
|
forgotten = yield make_deferred_yieldable(defer.gatherResults([
|
||||||
defer.maybeDeferred(
|
defer.maybeDeferred(
|
||||||
preserve_fn(store.who_forgot_in_room),
|
preserve_fn(store.who_forgot_in_room),
|
||||||
room_id,
|
room_id,
|
||||||
|
|
Loading…
Reference in a new issue