diff --git a/synapse/__init__.py b/synapse/__init__.py index 9ae406fd35..352309b66e 100644 --- a/synapse/__init__.py +++ b/synapse/__init__.py @@ -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() diff --git a/synapse/app/phone_stats_home.py b/synapse/app/phone_stats_home.py index 052296338d..64d5411a85 100644 --- a/synapse/app/phone_stats_home.py +++ b/synapse/app/phone_stats_home.py @@ -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] = {} diff --git a/synapse/handlers/worker_lock.py b/synapse/handlers/worker_lock.py index 7d2cfe1e91..51454aecaf 100644 --- a/synapse/handlers/worker_lock.py +++ b/synapse/handlers/worker_lock.py @@ -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 diff --git a/synapse/http/replicationagent.py b/synapse/http/replicationagent.py index f2eb58846e..003ce37c9a 100644 --- a/synapse/http/replicationagent.py +++ b/synapse/http/replicationagent.py @@ -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. diff --git a/synapse/logging/context.py b/synapse/logging/context.py index 92af52ef5c..da1810f9a3 100644 --- a/synapse/logging/context.py +++ b/synapse/logging/context.py @@ -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. diff --git a/synapse/metrics/background_process_metrics.py b/synapse/metrics/background_process_metrics.py index 5abdd107c4..8927a74f06 100644 --- a/synapse/metrics/background_process_metrics.py +++ b/synapse/metrics/background_process_metrics.py @@ -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") diff --git a/synapse/module_api/__init__.py b/synapse/module_api/__init__.py index 09d7939114..c79aefc6a9 100644 --- a/synapse/module_api/__init__.py +++ b/synapse/module_api/__init__.py @@ -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 diff --git a/synapse/storage/databases/main/lock.py b/synapse/storage/databases/main/lock.py index d2842e55e2..f63fa9ab02 100644 --- a/synapse/storage/databases/main/lock.py +++ b/synapse/storage/databases/main/lock.py @@ -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 diff --git a/synapse/storage/databases/main/push_rule.py b/synapse/storage/databases/main/push_rule.py index db26622c4d..c0f692dab5 100644 --- a/synapse/storage/databases/main/push_rule.py +++ b/synapse/storage/databases/main/push_rule.py @@ -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", diff --git a/synapse/util/async_helpers.py b/synapse/util/async_helpers.py index 4b846e9956..1425e540c2 100644 --- a/synapse/util/async_helpers.py +++ b/synapse/util/async_helpers.py @@ -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] diff --git a/synapse/util/batching_queue.py b/synapse/util/batching_queue.py index 67ecbd6f52..b9e603a7df 100644 --- a/synapse/util/batching_queue.py +++ b/synapse/util/batching_queue.py @@ -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 diff --git a/synapse/util/caches/expiringcache.py b/synapse/util/caches/expiringcache.py index fbf8e04263..107bc08c66 100644 --- a/synapse/util/caches/expiringcache.py +++ b/synapse/util/caches/expiringcache.py @@ -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)) diff --git a/synapse/util/caches/lrucache.py b/synapse/util/caches/lrucache.py index 8fcce0ebc5..6a680393c4 100644 --- a/synapse/util/caches/lrucache.py +++ b/synapse/util/caches/lrucache.py @@ -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. """ diff --git a/synapse/util/caches/response_cache.py b/synapse/util/caches/response_cache.py index 74930bf3d9..aaa1cd326e 100644 --- a/synapse/util/caches/response_cache.py +++ b/synapse/util/caches/response_cache.py @@ -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: diff --git a/synapse/util/distributor.py b/synapse/util/distributor.py index db3f1919af..bd91fb4fe0 100644 --- a/synapse/util/distributor.py +++ b/synapse/util/distributor.py @@ -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 "" % (self.name,) diff --git a/synapse/util/patch_inline_callbacks.py b/synapse/util/patch_inline_callbacks.py deleted file mode 100644 index 87dbd02e37..0000000000 --- a/synapse/util/patch_inline_callbacks.py +++ /dev/null @@ -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: -# . -# -# Originally licensed under the Apache License, Version 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 diff --git a/synapse/util/ratelimitutils.py b/synapse/util/ratelimitutils.py index 36464556ed..311299c26a 100644 --- a/synapse/util/ratelimitutils.py +++ b/synapse/util/ratelimitutils.py @@ -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 diff --git a/tests/test_utils/__init__.py b/tests/test_utils/__init__.py index 5ffb1bd8c8..9f4d1f0aec 100644 --- a/tests/test_utils/__init__.py +++ b/tests/test_utils/__init__.py @@ -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: """ diff --git a/tests/util/test_async_helpers.py b/tests/util/test_async_helpers.py index f415552746..4aa41a1ca7 100644 --- a/tests/util/test_async_helpers.py +++ b/tests/util/test_async_helpers.py @@ -44,7 +44,6 @@ from synapse.util.async_helpers import ( concurrently_execute, delay_cancellation, gather_optional_coroutines, - stop_cancellation, timeout_deferred, )