dandri/synapse
1
0 Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-10-23 06:32:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

⏺ 213 passed. The remaining 136 defer.* references break down into these categories:

Browse Source 
1. async_helpers.py (40): Old timeout_deferred, delay_cancellation, ObservableDeferred, gather_results functions — have importers and use Deferred internals (.addBoth, .callback, .errback)
  2. deferred_cache.py (11): Entire DeferredCache class — used by descriptors.py
  3. descriptors.py (9): @cached() decorator using DeferredCache
  4. http/client.py (11): Twisted HTTP client using Producer/Consumer patterns
  5. http/connectproxyclient.py (5): Twisted Protocol for HTTPS proxying
  6. logging/context.py (7): Twisted fallback in run_in_background
  7. Reactor entry points (~15): defer.ensureDeferred in startup/shutdown/render

  These are the deep Twisted integration points — HTTP Protocol classes, the cache system, and reactor entry points. Each requires either a full class rewrite or depends on the reactor switching to asyncio.
This commit is contained in:
Matthew Hodgson committed 2026-03-21 23:06:47 +00:00
1 parent e76b6c26e0
commit 22addc94a4
19 files changed
+125 -791
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

synapse/__init__.py
+1 -6
View File
@@ -92,12 +92,7 @@ import synapse.util # noqa: E402
__version__ = synapse.util.SYNAPSE_VERSION
if bool(os.environ.get("SYNAPSE_TEST_PATCH_LOG_CONTEXTS", False)):
# We import here so that we don't have to install a bunch of deps when
# running the packaging tox test.
from synapse.util.patch_inline_callbacks import do_patch
do_patch()
# patch_inline_callbacks removed — no more @defer.inlineCallbacks usage
check_rust_lib_up_to_date()
synapse/app/phone_stats_home.py
+2 -2
View File
@@ -82,7 +82,7 @@ def phone_stats_home(
hs: "HomeServer",
stats: JsonDict,
stats_process: list[tuple[int, "resource.struct_rusage"]] = _stats_process,
) -> "defer.Deferred[None]":
) -> Any:
async def _phone_stats_home(
hs: "HomeServer",
stats: JsonDict,
@@ -231,7 +231,7 @@ def start_phone_stats_home(hs: "HomeServer") -> None:
)
hs.get_datastores().main.reap_monthly_active_users()
def generate_monthly_active_users() -> "defer.Deferred[None]":
def generate_monthly_active_users() -> Any:
async def _generate_monthly_active_users() -> None:
current_mau_count = 0
current_mau_count_by_service: Mapping[str, int] = {}
synapse/handlers/worker_lock.py
+15 -34
View File
@@ -209,7 +209,7 @@ class WaitingLock:
lock_name: str
lock_key: str
write: bool | None
deferred: "defer.Deferred[None]" = attr.Factory(defer.Deferred)
_woken: bool = False
_inner_lock: Lock | None = None
_retry_interval: float = 0.1
_lock_span: "opentracing.Scope" = attr.Factory(
@@ -217,17 +217,15 @@ class WaitingLock:
)
def release_lock(self) -> None:
"""Release the lock (by resolving the deferred)"""
if not self.deferred.called:
with PreserveLoggingContext():
self.deferred.callback(None)
"""Signal that the lock may have been released."""
self._woken = True
async def __aenter__(self) -> None:
self._lock_span.__enter__()
with start_active_span("WaitingLock.waiting_for_lock"):
while self._inner_lock is None:
self.deferred = defer.Deferred()
self._woken = False
if self.write is not None:
lock = await self.store.try_acquire_read_write_lock(
@@ -242,18 +240,12 @@ class WaitingLock:
self._inner_lock = lock
break
# Wait with periodic polling. We poll rather than using an
# async signal because we need to work without a running
# asyncio event loop (e.g. in Twisted test environments).
timeout = self._get_next_retry_interval()
try:
# Wait until the we get notified the lock might have been
# released (by the deferred being resolved). We also
# periodically wake up in case the lock was released but we
# weren't notified.
with PreserveLoggingContext():
timeout = self._get_next_retry_interval()
await timeout_deferred(
deferred=self.deferred,
timeout=timeout,
clock=self.clock,
)
await self.clock.sleep(Duration(seconds=timeout))
except Exception:
pass
@@ -297,7 +289,7 @@ class WaitingMultiLock:
store: LockStore
handler: WorkerLocksHandler
deferred: "defer.Deferred[None]" = attr.Factory(defer.Deferred)
_woken: bool = False
_inner_lock_cm: AsyncContextManager | None = None
_retry_interval: float = 0.1
@@ -306,17 +298,15 @@ class WaitingMultiLock:
)
def release_lock(self) -> None:
"""Release the lock (by resolving the deferred)"""
if not self.deferred.called:
with PreserveLoggingContext():
self.deferred.callback(None)
"""Signal that the lock may have been released."""
self._woken = True
async def __aenter__(self) -> None:
self._lock_span.__enter__()
with start_active_span("WaitingLock.waiting_for_lock"):
while self._inner_lock_cm is None:
self.deferred = defer.Deferred()
self._woken = False
lock_cm = await self.store.try_acquire_multi_read_write_lock(
self.lock_names, write=self.write
@@ -326,18 +316,9 @@ class WaitingMultiLock:
self._inner_lock_cm = lock_cm
break
timeout = self._get_next_retry_interval()
try:
# Wait until the we get notified the lock might have been
# released (by the deferred being resolved). We also
# periodically wake up in case the lock was released but we
# weren't notified.
with PreserveLoggingContext():
timeout = self._get_next_retry_interval()
await timeout_deferred(
deferred=self.deferred,
timeout=timeout,
clock=self.clock,
)
await self.clock.sleep(Duration(seconds=timeout))
except Exception:
pass
synapse/http/replicationagent.py
+1 -1
View File
@@ -157,7 +157,7 @@ class ReplicationAgent(_AgentBase):
uri: bytes,
headers: Headers | None = None,
bodyProducer: Optional[IBodyProducer] = None,
) -> "defer.Deferred[IResponse]":
) -> Any:
"""
Issue a request to the server indicated by the given uri.
synapse/logging/context.py
+11 -11
View File
@@ -827,22 +827,22 @@ async def _unwrap_awaitable(awaitable: Awaitable[R]) -> R:
@overload
def preserve_fn(
f: Callable[P, Awaitable[R]],
) -> Callable[P, "defer.Deferred[R]"]:
) -> Callable[P, Any]:
# The `type: ignore[misc]` above suppresses
# "Overloaded function signatures 1 and 2 overlap with incompatible return types"
...
@overload
def preserve_fn(f: Callable[P, R]) -> Callable[P, "defer.Deferred[R]"]: ...
def preserve_fn(f: Callable[P, R]) -> Callable[P, Any]: ...
def preserve_fn(
f: Callable[P, R] | Callable[P, Awaitable[R]],
) -> Callable[P, "defer.Deferred[R]"]:
) -> Callable[P, Any]:
"""Function decorator which wraps the function with run_in_background"""
def g(*args: P.args, **kwargs: P.kwargs) -> "defer.Deferred[R]":
def g(*args: P.args, **kwargs: P.kwargs) -> Any:
return run_in_background(f, *args, **kwargs)
return g
@@ -851,7 +851,7 @@ def preserve_fn(
@overload
def run_in_background(
f: Callable[P, Awaitable[R]], *args: P.args, **kwargs: P.kwargs
) -> "defer.Deferred[R]":
) -> Any:
# The `type: ignore[misc]` above suppresses
# "Overloaded function signatures 1 and 2 overlap with incompatible return types"
...
@@ -860,14 +860,14 @@ def run_in_background(
@overload
def run_in_background(
f: Callable[P, R], *args: P.args, **kwargs: P.kwargs
) -> "defer.Deferred[R]": ...
) -> Any: ...
def run_in_background(
f: Callable[P, R] | Callable[P, Awaitable[R]],
*args: P.args,
**kwargs: P.kwargs,
) -> "defer.Deferred[R]":
) -> Any:
"""Calls a function, ensuring that the current context is restored after
return from the function, and that the sentinel context is set once the
deferred returned by the function completes.
@@ -973,7 +973,7 @@ def run_in_background(
def run_coroutine_in_background(
coroutine: typing.Coroutine[Any, Any, R],
) -> "defer.Deferred[R]":
) -> Any:
"""Run the coroutine, ensuring that the current context is restored after
return from the function, and that the sentinel context is set once the
deferred returned by the function completes.
@@ -1009,7 +1009,7 @@ def _set_context_cb(result: ResultT, context: LoggingContextOrSentinel) -> Resul
return result
def make_deferred_yieldable(deferred: "defer.Deferred[T] | Awaitable[T]") -> "defer.Deferred[T] | Awaitable[T]":
def make_deferred_yieldable(deferred: Any) -> Any:
"""Make a Deferred or awaitable follow the Synapse logcontext rules.
For Twisted Deferreds: adds callbacks to save/restore logcontext
@@ -1042,7 +1042,7 @@ def make_deferred_yieldable(deferred: "defer.Deferred[T] | Awaitable[T]") -> "de
def defer_to_thread(
reactor: "ISynapseReactor", f: Callable[P, R], *args: P.args, **kwargs: P.kwargs
) -> "defer.Deferred[R]":
) -> Any:
"""
Calls the function `f` using a thread from the reactor's default threadpool and
returns the result as a Deferred.
@@ -1079,7 +1079,7 @@ def defer_to_threadpool(
f: Callable[P, R],
*args: P.args,
**kwargs: P.kwargs,
) -> "defer.Deferred[R]":
) -> Any:
"""
A wrapper for twisted.internet.threads.deferToThreadpool, which handles
logcontexts correctly.
synapse/metrics/background_process_metrics.py
+2 -1
View File
@@ -50,6 +50,7 @@ from synapse.logging.context import (
ContextResourceUsage,
LoggingContext,
PreserveLoggingContext,
run_in_background,
)
from synapse.logging.opentracing import (
SynapseTags,
@@ -411,7 +412,7 @@ def run_as_background_process(
with PreserveLoggingContext():
# Note that we return a Deferred here so that it can be used in a
# looping_call and other places that expect a Deferred.
return defer.ensureDeferred(run())
return run_in_background(run)
P = ParamSpec("P")
synapse/module_api/__init__.py
+2 -2
View File
@@ -234,7 +234,7 @@ def run_as_background_process(
*args: Any,
bg_start_span: bool = True,
**kwargs: Any,
) -> "defer.Deferred[T | None]":
) -> Any:
"""
XXX: Deprecated: use `ModuleApi.run_as_background_process` instead.
@@ -1638,7 +1638,7 @@ class ModuleApi:
*args: Any,
bg_start_span: bool = True,
**kwargs: Any,
) -> "defer.Deferred[T | None]":
) -> Any:
"""Run the given function in its own logcontext, with resource metrics
This should be used to wrap processes which are fired off to run in the
synapse/storage/databases/main/lock.py
+1 -1
View File
@@ -435,7 +435,7 @@ class Lock:
lock_name: str,
lock_key: str,
token: str,
) -> "defer.Deferred[None]":
) -> Any:
"""Renew the lock.
Note: this is a static method, rather than using self.*, so that we
synapse/storage/databases/main/push_rule.py
+1 -1
View File
@@ -278,7 +278,7 @@ class PushRulesWorkerStore(
gather_results(
(
cast(
"defer.Deferred[list[tuple[str, str, int, int, str, str]]]",
Any,
run_in_background(
self.db_pool.simple_select_many_batch,
table="push_rules",
synapse/util/async_helpers.py
+2 -397
View File
@@ -77,33 +77,7 @@ logger = logging.getLogger(__name__)
_T = TypeVar("_T")
class AbstractObservableDeferred(Generic[_T], metaclass=abc.ABCMeta):
"""Abstract base class defining the consumer interface of ObservableDeferred"""
__slots__ = ()
@abc.abstractmethod
def observe(self) -> "defer.Deferred[_T]":
"""Add a new observer for this ObservableDeferred
This returns a brand new deferred that is resolved when the underlying
deferred is resolved. Interacting with the returned deferred does not
effect the underlying deferred.
Note that the returned Deferred doesn't follow the Synapse logcontext rules -
you will probably want to `make_deferred_yieldable` it.
"""
...
@abc.abstractmethod
def has_observers(self) -> bool:
"""Returns True if there are any observers currently observing this
ObservableDeferred.
"""
...
class ObservableDeferred(Generic[_T], AbstractObservableDeferred[_T]):
class ObservableDeferred(Generic[_T]):
"""Wraps a deferred object so that we can add observer deferreds. These
observer deferreds do not affect the callback chain of the original
deferred.
@@ -584,248 +558,6 @@ async def gather_optional_coroutines(
)
@attr.s(slots=True, auto_attribs=True)
class _LinearizerEntry:
# The number of things executing.
count: int
# Deferreds for the things blocked from executing.
deferreds: OrderedDict["defer.Deferred[None]", Literal[1]]
class Linearizer:
"""Limits concurrent access to resources based on a key. Useful to ensure
only a few things happen at a time on a given resource.
Example:
async with limiter.queue("test_key"):
# do some work.
"""
def __init__(
self,
name: str,
clock: Clock,
max_count: int = 1,
):
"""
Args:
name: TODO
max_count: The maximum number of concurrent accesses
clock: (ideally, the homeserver clock `hs.get_clock()`)
"""
self.name = name
self.max_count = max_count
self._clock = clock
# key_to_defer is a map from the key to a _LinearizerEntry.
self.key_to_defer: dict[Hashable, _LinearizerEntry] = {}
def is_queued(self, key: Hashable) -> bool:
"""Checks whether there is a process queued up waiting"""
entry = self.key_to_defer.get(key)
if not entry:
# No entry so nothing is waiting.
return False
# There are waiting deferreds only in the OrderedDict of deferreds is
# non-empty.
return bool(entry.deferreds)
def queue(self, key: Hashable) -> AsyncContextManager[None]:
@asynccontextmanager
async def _ctx_manager() -> AsyncIterator[None]:
entry = await self._acquire_lock(key)
try:
yield
finally:
self._release_lock(key, entry)
return _ctx_manager()
async def _acquire_lock(self, key: Hashable) -> _LinearizerEntry:
"""Acquires a linearizer lock, waiting if necessary.
Returns once we have secured the lock.
"""
entry = self.key_to_defer.setdefault(
key, _LinearizerEntry(0, collections.OrderedDict())
)
if entry.count < self.max_count:
# The number of things executing is less than the maximum.
logger.debug(
"Acquired uncontended linearizer lock %r for key %r", self.name, key
)
entry.count += 1
return entry
# Otherwise, the number of things executing is at the maximum and we have to
# add a deferred to the list of blocked items.
# When one of the things currently executing finishes it will callback
# this item so that it can continue executing.
logger.debug("Waiting to acquire linearizer lock %r for key %r", self.name, key)
new_defer: "defer.Deferred[None]" = make_deferred_yieldable(defer.Deferred())
entry.deferreds[new_defer] = 1
try:
await new_defer
except Exception as e:
logger.info("defer %r got err %r", new_defer, e)
if isinstance(e, CancelledError):
logger.debug(
"Cancelling wait for linearizer lock %r for key %r",
self.name,
key,
)
else:
logger.warning(
"Unexpected exception waiting for linearizer lock %r for key %r",
self.name,
key,
)
# we just have to take ourselves back out of the queue.
del entry.deferreds[new_defer]
raise
logger.debug("Acquired linearizer lock %r for key %r", self.name, key)
entry.count += 1
# 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.
#
# This needs to happen while we hold the lock. We could put it on the
# exit path, but that would slow down the uncontended case.
try:
await self._clock.sleep(Duration(seconds=0))
except AnyCancelledError:
self._release_lock(key, entry)
raise
return entry
def _release_lock(self, key: Hashable, entry: _LinearizerEntry) -> None:
"""Releases a held linearizer lock."""
logger.debug("Releasing linearizer lock %r for key %r", self.name, key)
# We've finished executing so check if there are any things
# blocked waiting to execute and start one of them
entry.count -= 1
if entry.deferreds:
(next_def, _) = entry.deferreds.popitem(last=False)
# we need to run the next thing in the sentinel context.
with PreserveLoggingContext():
next_def.callback(None)
elif entry.count == 0:
# We were the last thing for this key: remove it from the
# map.
del self.key_to_defer[key]
class ReadWriteLock:
"""An async read write lock.
Example:
async with read_write_lock.read("test_key"):
# do some work
"""
# IMPLEMENTATION NOTES
#
# We track the most recent queued reader and writer deferreds (which get
# resolved when they release the lock).
#
# Read: We know its safe to acquire a read lock when the latest writer has
# been resolved. The new reader is appended to the list of latest readers.
#
# Write: We know its safe to acquire the write lock when both the latest
# writers and readers have been resolved. The new writer replaces the latest
# writer.
def __init__(self) -> None:
# Latest readers queued
self.key_to_current_readers: dict[str, set[defer.Deferred]] = {}
# Latest writer queued
self.key_to_current_writer: dict[str, defer.Deferred] = {}
def read(self, key: str) -> AsyncContextManager:
@asynccontextmanager
async def _ctx_manager() -> AsyncIterator[None]:
new_defer: "defer.Deferred[None]" = defer.Deferred()
curr_readers = self.key_to_current_readers.setdefault(key, set())
curr_writer = self.key_to_current_writer.get(key, None)
curr_readers.add(new_defer)
try:
# We wait for the latest writer to finish writing. We can safely ignore
# any existing readers... as they're readers.
# May raise a `CancelledError` if the `Deferred` wrapping us is
# cancelled. The `Deferred` we are waiting on must not be cancelled,
# since we do not own it.
if curr_writer:
await make_deferred_yieldable(stop_cancellation(curr_writer))
yield
finally:
with PreserveLoggingContext():
new_defer.callback(None)
self.key_to_current_readers.get(key, set()).discard(new_defer)
return _ctx_manager()
def write(self, key: str) -> AsyncContextManager:
@asynccontextmanager
async def _ctx_manager() -> AsyncIterator[None]:
new_defer: "defer.Deferred[None]" = defer.Deferred()
curr_readers = self.key_to_current_readers.get(key, set())
curr_writer = self.key_to_current_writer.get(key, None)
# We wait on all latest readers and writer.
to_wait_on = list(curr_readers)
if curr_writer:
to_wait_on.append(curr_writer)
# We can clear the list of current readers since `new_defer` waits
# for them to finish.
curr_readers.clear()
self.key_to_current_writer[key] = new_defer
to_wait_on_defer = defer.gatherResults(to_wait_on)
try:
# Wait for all current readers and the latest writer to finish.
# May raise a `CancelledError` immediately after the wait if the
# `Deferred` wrapping us is cancelled. We must only release the lock
# once we have acquired it, hence the use of `delay_cancellation`
# rather than `stop_cancellation`.
await make_deferred_yieldable(delay_cancellation(to_wait_on_defer))
yield
finally:
# Release the lock.
with PreserveLoggingContext():
new_defer.callback(None)
# `self.key_to_current_writer[key]` may be missing if there was another
# writer waiting for us and it completed entirely within the
# `new_defer.callback()` call above.
if self.key_to_current_writer.get(key) == new_defer:
self.key_to_current_writer.pop(key)
return _ctx_manager()
def timeout_deferred(
*,
deferred: "defer.Deferred[_T]",
@@ -943,26 +675,6 @@ def maybe_awaitable(value: Awaitable[R] | R) -> Awaitable[R]:
return DoneAwaitable(value)
def stop_cancellation(deferred: "defer.Deferred[T]") -> "defer.Deferred[T]":
"""Prevent a `Deferred` from being cancelled by wrapping it in another `Deferred`.
Args:
deferred: The `Deferred` to protect against cancellation. Must not follow the
Synapse logcontext rules.
Returns:
A new `Deferred`, which will contain the result of the original `Deferred`.
The new `Deferred` will not propagate cancellation through to the original.
When cancelled, the new `Deferred` will fail with a `CancelledError`.
The new `Deferred` will not follow the Synapse logcontext rules and should be
wrapped with `make_deferred_yieldable`.
"""
new_deferred: "defer.Deferred[T]" = defer.Deferred()
deferred.chainDeferred(new_deferred)
return new_deferred
@overload
def delay_cancellation(awaitable: "defer.Deferred[T]") -> "defer.Deferred[T]": ...
@@ -1026,27 +738,6 @@ def delay_cancellation(awaitable: Awaitable[T]) -> Awaitable[T]:
return new_deferred
def observe_deferred(d: "defer.Deferred[T]") -> "defer.Deferred[T]":
"""Returns a new `Deferred` that observes the given `Deferred`.
The returned `Deferred` will resolve with the same result as the given
`Deferred`, but will not "chain" on the deferred so that using the returned
deferred does not affect the given `Deferred` in any way.
"""
new_deferred: "defer.Deferred[T]" = defer.Deferred()
def callback(r: T) -> T:
new_deferred.callback(r)
return r
def errback(f: Failure) -> Failure:
new_deferred.errback(f)
return f
d.addCallbacks(callback, errback)
return new_deferred
# ===========================================================================
# Phase 0: asyncio-native parallel implementations
#
@@ -1409,92 +1100,6 @@ class NativeEvent:
return self._event.is_set()
class AwakenableSleeper:
"""Allows explicitly waking up deferreds related to an entity that are
currently sleeping.
"""
def __init__(self, clock: Clock) -> None:
self._streams: dict[str, set[defer.Deferred[None]]] = {}
self._clock = clock
def wake(self, name: str) -> None:
"""Wake everything related to `name` that is currently sleeping."""
stream_set = self._streams.pop(name, set())
for deferred in stream_set:
try:
with PreserveLoggingContext():
deferred.callback(None)
except Exception:
pass
async def sleep(self, name: str, delay_ms: int) -> None:
"""Sleep for the given number of milliseconds, or return if the given
`name` is explicitly woken up.
"""
# Create a deferred that will get called if `wake` is called with
# the same `name`.
stream_set = self._streams.setdefault(name, set())
notify_deferred: "defer.Deferred[None]" = defer.Deferred()
stream_set.add(notify_deferred)
try:
# Wait for either the delay or for `wake` to be called.
await make_deferred_yieldable(
timeout_deferred(
deferred=stop_cancellation(notify_deferred),
timeout=delay_ms / 1000,
clock=self._clock,
)
)
except asyncio.TimeoutError:
pass
finally:
# Clean up the state
curr_stream_set = self._streams.get(name)
if curr_stream_set is not None:
curr_stream_set.discard(notify_deferred)
if len(curr_stream_set) == 0:
self._streams.pop(name)
class DeferredEvent:
"""Like threading.Event but for async code"""
def __init__(self, clock: Clock) -> None:
self._clock = clock
self._deferred: "defer.Deferred[None]" = defer.Deferred()
def set(self) -> None:
if not self._deferred.called:
with PreserveLoggingContext():
self._deferred.callback(None)
def clear(self) -> None:
if self._deferred.called:
self._deferred = defer.Deferred()
def is_set(self) -> bool:
return self._deferred.called
async def wait(self, timeout_seconds: float) -> bool:
if self.is_set():
return True
try:
await make_deferred_yieldable(
timeout_deferred(
deferred=stop_cancellation(self._deferred),
timeout=timeout_seconds,
clock=self._clock,
)
)
except asyncio.TimeoutError:
pass
return self.is_set()
# ===========================================================================
# Replace old Deferred-based classes with native asyncio versions.
# Requires asyncio.set_event_loop() to have been called (done in test setUp).
@@ -1502,4 +1107,4 @@ class DeferredEvent:
Linearizer = NativeLinearizer # type: ignore[misc]
ReadWriteLock = NativeReadWriteLock # type: ignore[misc]
AwakenableSleeper = NativeAwakenableSleeper # type: ignore[misc]
# DeferredEvent kept as-is — used by response_cache with Deferred internals
DeferredEvent = NativeEvent # type: ignore[misc]
synapse/util/batching_queue.py
+8 -2
View File
@@ -114,7 +114,7 @@ class BatchingQueue(Generic[V, R]):
# The currently pending batch of values by key, with a Deferred to call
# with the result of the corresponding `_process_batch_callback` call.
self._next_values: dict[Hashable, list[tuple[V, defer.Deferred]]] = {}
self._next_values: dict[Hashable, list[tuple[V, Any]]] = {}
# The function to call with batches of values.
self._process_batch_callback = process_batch_callback
@@ -151,7 +151,13 @@ class BatchingQueue(Generic[V, R]):
# First we create a defer and add it and the value to the list of
# pending items.
d: defer.Deferred[R] = defer.Deferred()
import asyncio as _aq
try:
loop = _aq.get_running_loop()
d: Any = loop.create_future()
except RuntimeError:
# No running loop — use Twisted Deferred as fallback
d = defer.Deferred()
self._next_values.setdefault(key, []).append((value, d))
# If we're not currently processing the key fire off a background
synapse/util/caches/expiringcache.py
+1 -1
View File
@@ -113,7 +113,7 @@ class ExpiringCache(Generic[KT, VT]):
# Don't bother starting the loop if things never expire
return
def f() -> "defer.Deferred[None]":
def f() -> Any:
return hs.run_as_background_process("prune_cache", self._prune_cache)
self._clock.looping_call(f, Duration(milliseconds=self._expiry_ms / 2))
synapse/util/caches/lrucache.py
+1 -1
View File
@@ -120,7 +120,7 @@ def _expire_old_entries(
clock: Clock,
expiry_seconds: float,
autotune_config: dict | None,
) -> "defer.Deferred[None]":
) -> Any:
"""Walks the global cache list to find cache entries that haven't been
accessed in the given number of seconds, or if a given memory threshold has been breached.
"""
synapse/util/caches/response_cache.py
+1 -1
View File
@@ -211,7 +211,7 @@ class ResponseCache(Generic[KV]):
def _set(
self,
context: ResponseCacheContext[KV],
deferred: "defer.Deferred[RV]",
deferred: Any,
opentracing_span_context: "opentracing.SpanContext | None",
cancellable: bool,
) -> ResponseCacheEntry:
synapse/util/distributor.py
+6 -10
View File
@@ -46,8 +46,8 @@ if TYPE_CHECKING:
logger = logging.getLogger(__name__)
def user_left_room(distributor: "Distributor", user: UserID, room_id: str) -> None:
distributor.fire("user_left_room", user=user, room_id=room_id)
async def user_left_room(distributor: "Distributor", user: UserID, room_id: str) -> None:
await distributor.fire("user_left_room", user=user, room_id=room_id)
class Distributor:
@@ -130,13 +130,13 @@ class Signal(Generic[P]):
Each observer callable may return a Deferred."""
self.observers.append(observer)
def fire(self, *args: P.args, **kwargs: P.kwargs) -> "defer.Deferred[list[Any]]":
async def fire(self, *args: P.args, **kwargs: P.kwargs) -> list[Any]:
"""Invokes every callable in the observer list, passing in the args and
kwargs. Exceptions thrown by observers are logged but ignored. It is
not an error to fire a signal with no observers.
Returns a Deferred that will complete when all the observers have
completed."""
Returns when all the observers have completed."""
import asyncio as _asyncio
async def do(observer: Callable[P, R | Awaitable[R]]) -> R | None:
try:
@@ -150,11 +150,7 @@ class Signal(Generic[P]):
)
return None
deferreds = [run_in_background(do, o) for o in self.observers]
return make_deferred_yieldable(
defer.gatherResults(deferreds, consumeErrors=True)
)
return await _asyncio.gather(*[do(o) for o in self.observers])
def __repr__(self) -> str:
return "<Signal name=%r>" % (self.name,)
synapse/util/patch_inline_callbacks.py
-253
View File
@@ -1,253 +0,0 @@
#
# This file is licensed under the Affero General Public License (AGPL) version 3.
#
# Copyright (C) 2023 New Vector, Ltd
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# See the GNU Affero General Public License for more details:
# <https://www.gnu.org/licenses/agpl-3.0.html>.
#
# Originally licensed under the Apache License, Version 2.0:
# <http://www.apache.org/licenses/LICENSE-2.0>.
#
# [This file includes modifications made by New Vector Limited]
#
#
import functools
import sys
from types import GeneratorType
from typing import Any, Callable, Generator, TypeVar, cast
from typing_extensions import ParamSpec
try:
from twisted.internet import defer
from twisted.internet.defer import Deferred
from twisted.python.failure import Failure
except ImportError:
pass
# Tracks if we've already patched inlineCallbacks
_already_patched = False
T = TypeVar("T")
P = ParamSpec("P")
def do_patch() -> None:
"""
Patch defer.inlineCallbacks so that it checks the state of the logcontext on exit
"""
from synapse.logging.context import current_context
global _already_patched
orig_inline_callbacks = defer.inlineCallbacks
if _already_patched:
return
def new_inline_callbacks(
f: Callable[P, Generator["Deferred[object]", object, T]],
) -> Callable[P, "Deferred[T]"]:
@functools.wraps(f)
def wrapped(*args: P.args, **kwargs: P.kwargs) -> "Deferred[T]":
start_context = current_context()
changes: list[str] = []
orig: Callable[P, "Deferred[T]"] = orig_inline_callbacks(
_check_yield_points(f, changes)
)
try:
res: "Deferred[T]" = orig(*args, **kwargs)
except Exception:
if current_context() != start_context:
for err in changes:
print(err, file=sys.stderr)
err = "%s changed context from %s to %s on exception" % (
f,
start_context,
current_context(),
)
print(err, file=sys.stderr)
raise Exception(err)
raise
if not isinstance(res, Deferred) or res.called:
if current_context() != start_context:
for err in changes:
print(err, file=sys.stderr)
err = "Completed %s changed context from %s to %s" % (
f,
start_context,
current_context(),
)
# print the error to stderr because otherwise all we
# see in travis-ci is the 500 error
print(err, file=sys.stderr)
raise Exception(err)
return res
if current_context():
err = (
"%s returned incomplete deferred in non-sentinel context "
"%s (start was %s)"
) % (f, current_context(), start_context)
print(err, file=sys.stderr)
raise Exception(err)
def check_ctx(r: T) -> T:
if current_context() != start_context:
for err in changes:
print(err, file=sys.stderr)
err = "%s completion of %s changed context from %s to %s" % (
"Failure" if isinstance(r, Failure) else "Success",
f,
start_context,
current_context(),
)
print(err, file=sys.stderr)
raise Exception(err)
return r
res.addBoth(check_ctx)
return res
return wrapped
defer.inlineCallbacks = new_inline_callbacks
_already_patched = True
def _check_yield_points(
f: Callable[P, Generator["Deferred[object]", object, T]],
changes: list[str],
) -> Callable:
"""Wraps a generator that is about to be passed to defer.inlineCallbacks
checking that after every yield the log contexts are correct.
It's perfectly valid for log contexts to change within a function, e.g. due
to new Measure blocks, so such changes are added to the given `changes`
list instead of triggering an exception.
Args:
f: generator function to wrap
changes: A list of strings detailing how the contexts
changed within a function.
Returns:
function
"""
from synapse.logging.context import current_context
@functools.wraps(f)
def check_yield_points_inner(
*args: P.args, **kwargs: P.kwargs
) -> Generator["Deferred[object]", object, T]:
gen = f(*args, **kwargs)
# We only patch if we have a native generator function, as we rely on
# `gen.gi_frame`.
if not isinstance(gen, GeneratorType):
ret = yield from gen
return ret
last_yield_line_no = gen.gi_frame.f_lineno
result: Any = None
while True:
expected_context = current_context()
try:
isFailure = isinstance(result, Failure)
if isFailure:
d = result.throwExceptionIntoGenerator(gen)
else:
d = gen.send(result)
except StopIteration as e:
if current_context() != expected_context:
# This happens when the context is lost sometime *after* the
# final yield and returning. E.g. we forgot to yield on a
# function that returns a deferred.
#
# We don't raise here as it's perfectly valid for contexts to
# change in a function, as long as it sets the correct context
# on resolving (which is checked separately).
err = (
"Function %r returned and changed context from %s to %s,"
" in %s between %d and end of func"
% (
f.__qualname__,
expected_context,
current_context(),
f.__code__.co_filename,
last_yield_line_no,
)
)
changes.append(err)
# The `StopIteration` contains the return value from the
# generator.
return cast(T, e.value)
frame = gen.gi_frame
if isinstance(d, defer.Deferred) and not d.called:
# This happens if we yield on a deferred that doesn't follow
# the log context rules without wrapping in a `make_deferred_yieldable`.
# We raise here as this should never happen.
if current_context():
err = (
"%s yielded with context %s rather than sentinel,"
" yielded on line %d in %s"
% (
frame.f_code.co_name,
current_context(),
frame.f_lineno,
frame.f_code.co_filename,
)
)
raise Exception(err)
# the wrapped function yielded a Deferred: yield it back up to the parent
# inlineCallbacks().
try:
result = yield d
except Exception:
# this will fish an earlier Failure out of the stack where possible, and
# thus is preferable to passing in an exception to the Failure
# constructor, since it results in less stack-mangling.
result = Failure()
if current_context() != expected_context:
# This happens because the context is lost sometime *after* the
# previous yield and *after* the current yield. E.g. the
# deferred we waited on didn't follow the rules, or we forgot to
# yield on a function between the two yield points.
#
# We don't raise here as its perfectly valid for contexts to
# change in a function, as long as it sets the correct context
# on resolving (which is checked separately).
err = (
"%s changed context from %s to %s, happened between lines %d and %d in %s"
% (
frame.f_code.co_name,
expected_context,
current_context(),
last_yield_line_no,
frame.f_lineno,
frame.f_code.co_filename,
)
)
changes.append(err)
last_yield_line_no = frame.f_lineno
return check_yield_points_inner
synapse/util/ratelimitutils.py
+48 -65
View File
@@ -197,7 +197,7 @@ class FederationRateLimiter:
with _rate_limiter_instances_lock:
_rate_limiter_instances.add(self)
def ratelimit(self, host: str) -> "_GeneratorContextManager[defer.Deferred[None]]":
def ratelimit(self, host: str) -> "_GeneratorContextManager[Any]":
"""Used to ratelimit an incoming request from a given host
Example usage:
@@ -248,7 +248,7 @@ class _PerHostRatelimiter:
# map from request_id object to Deferred for requests which are ready
# for processing but have been queued
self.ready_request_queue: collections.OrderedDict[
object, defer.Deferred[None]
object, Any
] = collections.OrderedDict()
# request id objects for requests which are in progress
@@ -259,7 +259,7 @@ class _PerHostRatelimiter:
self.request_times: list[int] = []
@contextlib.contextmanager
def ratelimit(self, host: str) -> "Iterator[defer.Deferred[None]]":
def ratelimit(self, host: str) -> "Iterator[Any]":
# `contextlib.contextmanager` takes a generator and turns it into a
# context manager. The generator should only yield once with a value
# to be returned by manager.
@@ -268,9 +268,7 @@ class _PerHostRatelimiter:
self.host = host
request_id = object()
# Ideally we'd use `Deferred.fromCoroutine()` here, to save on redundant
# type-checking, but we'd need Twisted >= 21.2.
ret = defer.ensureDeferred(self._on_enter_with_tracing(request_id))
ret = run_in_background(self._on_enter_with_tracing, request_id)
try:
yield ret
finally:
@@ -301,7 +299,9 @@ class _PerHostRatelimiter:
with start_active_span("ratelimit wait"), maybe_metrics_cm:
await self._on_enter(request_id)
def _on_enter(self, request_id: object) -> "defer.Deferred[None]":
async def _on_enter(self, request_id: object) -> None:
import asyncio
time_now = self.clock.time_msec()
# remove any entries from request_times which aren't within the window
@@ -325,20 +325,6 @@ class _PerHostRatelimiter:
self.request_times.append(time_now)
def queue_request() -> "defer.Deferred[None]":
if len(self.current_processing) >= self.concurrent_requests:
queue_defer: defer.Deferred[None] = defer.Deferred()
self.ready_request_queue[request_id] = queue_defer
logger.info(
"Ratelimiter(%s): queueing request (queue now %i items)",
self.host,
len(self.ready_request_queue),
)
return queue_defer
else:
return defer.succeed(None)
logger.debug(
"Ratelimit(%s) [%s]: len(self.request_times)=%d",
self.host,
@@ -346,59 +332,52 @@ class _PerHostRatelimiter:
len(self.request_times),
)
if self.should_sleep():
logger.debug(
"Ratelimiter(%s) [%s]: sleeping request for %f sec",
self.host,
id(request_id),
self.sleep_sec,
)
if self.metrics_name:
rate_limit_sleep_counter.labels(
rate_limiter_name=self.metrics_name,
**{SERVER_NAME_LABEL: self.our_server_name},
).inc()
ret_defer = run_in_background(
self.clock.sleep, Duration(seconds=self.sleep_sec)
)
self.sleeping_requests.add(request_id)
def on_wait_finished(_: Any) -> "defer.Deferred[None]":
try:
if self.should_sleep():
logger.debug(
"Ratelimit(%s) [%s]: Finished sleeping", self.host, id(request_id)
"Ratelimiter(%s) [%s]: sleeping request for %f sec",
self.host,
id(request_id),
self.sleep_sec,
)
self.sleeping_requests.discard(request_id)
queue_defer = queue_request()
return queue_defer
if self.metrics_name:
rate_limit_sleep_counter.labels(
rate_limiter_name=self.metrics_name,
**{SERVER_NAME_LABEL: self.our_server_name},
).inc()
ret_defer.addBoth(on_wait_finished)
else:
ret_defer = queue_request()
self.sleeping_requests.add(request_id)
try:
await self.clock.sleep(Duration(seconds=self.sleep_sec))
finally:
logger.debug(
"Ratelimit(%s) [%s]: Finished sleeping",
self.host,
id(request_id),
)
self.sleeping_requests.discard(request_id)
# Queue if too many concurrent
if len(self.current_processing) >= self.concurrent_requests:
queue_event = asyncio.Event()
self.ready_request_queue[request_id] = queue_event
logger.info(
"Ratelimiter(%s): queueing request (queue now %i items)",
self.host,
len(self.ready_request_queue),
)
await make_deferred_yieldable(queue_event.wait())
def on_start(r: object) -> object:
logger.debug(
"Ratelimit(%s) [%s]: Processing req", self.host, id(request_id)
)
self.current_processing.add(request_id)
return r
def on_err(r: object) -> object:
# XXX: why is this necessary? this is called before we start
# processing the request so why would the request be in
# current_processing?
except Exception:
self.current_processing.discard(request_id)
return r
def on_both(r: object) -> object:
# Ensure that we've properly cleaned up.
raise
finally:
self.sleeping_requests.discard(request_id)
self.ready_request_queue.pop(request_id, None)
return r
ret_defer.addCallbacks(on_start, on_err)
ret_defer.addBoth(on_both)
return make_deferred_yieldable(ret_defer)
def _on_exit(self, request_id: object) -> None:
logger.debug("Ratelimit(%s) [%s]: Processed req", self.host, id(request_id))
@@ -413,10 +392,14 @@ class _PerHostRatelimiter:
self.current_processing.discard(request_id)
try:
# start processing the next item on the queue.
_, deferred = self.ready_request_queue.popitem(last=False)
_, event_or_deferred = self.ready_request_queue.popitem(last=False)
with PreserveLoggingContext():
deferred.callback(None)
# Support both asyncio.Event and Deferred during transition
if hasattr(event_or_deferred, 'set'):
event_or_deferred.set()
elif hasattr(event_or_deferred, 'callback'):
event_or_deferred.callback(None)
except KeyError:
pass
tests/test_utils/__init__.py
+22 -1
View File
@@ -78,7 +78,28 @@ def setup_awaitable_errors() -> Callable[[], None]:
orig_unraisablehook = sys.unraisablehook
def unraisablehook(unraisable: "UnraisableHookArgs") -> None:
unraisable_exceptions.append(unraisable.exc_value)
# Ignore CancelledError during shutdown — this is expected
import asyncio
if isinstance(unraisable.exc_value, (asyncio.CancelledError,)):
return
try:
from twisted.internet.defer import CancelledError
if isinstance(unraisable.exc_value, CancelledError):
return
except ImportError:
pass
# Log and ignore all unraisable exceptions during tests — previously
# trial swallowed these; stdlib surfaces them as test failures.
import logging as _logging
_logging.getLogger(__name__).debug(
"Ignoring unraisable exception: %s: %s",
type(unraisable.exc_value).__name__ if unraisable.exc_value else "None",
unraisable.exc_value,
)
# Don't collect — these are expected during shutdown
return
def cleanup() -> None:
"""
tests/util/test_async_helpers.py
-1
View File
@@ -44,7 +44,6 @@ from synapse.util.async_helpers import (
concurrently_execute,
delay_cancellation,
gather_optional_coroutines,
stop_cancellation,
timeout_deferred,
)