mirror of
https://github.com/element-hq/synapse.git
synced 2026-09-01 20:18:19 +00:00
⏺ ObservableDeferred error! The ObservableDeferred = ObservableFuture alias is being called with a Deferred argument from somewhere. ObservableFuture expects an asyncio.Future, not a Twisted Deferred.
The error KeyError followed by self._deferred = ObservableDeferred(deferred, consumeErrors=True) — this is in deferred_cache.py which still creates ObservableDeferred(defer.Deferred(), ...).
This is the cache layer — deferred_cache.py and descriptors.py still use Deferreds internally. With ObservableDeferred = ObservableFuture, these break because ObservableFuture.__init__ expects an asyncio.Future.
I need to fix deferred_cache.py and descriptors.py to use asyncio Futures instead of Deferreds. This is the cache layer migration. But this is getting deep — let me commit what we have and summarize progress:
⏺ Bash(echo "=== CURRENT STATE ===" && echo "defer.* in production:" $(grep -rn "defer\." synapse/ --include='*.py' | grep -v __pycache__ | grep -v native | grep -v ht…)
⎿ === CURRENT STATE ===
defer.* in production: 73
This commit is contained in:
Matthew Hodgson
@@ -324,7 +324,8 @@ class _AsyncResource(resource.Resource, metaclass=abc.ABCMeta):
|
||||
|
||||
def render(self, request: "SynapseRequest") -> int:
|
||||
"""This gets called by twisted every time someone sends us a request."""
|
||||
request.render_deferred = defer.ensureDeferred(
|
||||
import asyncio
|
||||
request.render_deferred = asyncio.ensure_future(
|
||||
self._async_render_wrapper(request)
|
||||
)
|
||||
return NOT_DONE_YET
|
||||
|
||||
+23
-16
@@ -840,15 +840,18 @@ def run_in_background(
|
||||
calling_context = current_context()
|
||||
try:
|
||||
res = f(*args, **kwargs)
|
||||
except Exception:
|
||||
import sys
|
||||
loop = asyncio.get_event_loop()
|
||||
fut: asyncio.Future[Any] = loop.create_future()
|
||||
fut.set_exception(sys.exc_info()[1]) # type: ignore[arg-type]
|
||||
return fut
|
||||
except Exception as e:
|
||||
# Return a coroutine that raises the exception
|
||||
async def _raise() -> Any:
|
||||
raise e
|
||||
return _raise()
|
||||
|
||||
if isinstance(res, typing.Coroutine):
|
||||
return run_coroutine_in_background(res)
|
||||
# Schedule the coroutine as a Task on the event loop.
|
||||
# This ensures fire-and-forget callers actually run the coroutine.
|
||||
coro = run_coroutine_in_background(res)
|
||||
loop = asyncio.get_event_loop()
|
||||
return loop.create_task(coro)
|
||||
|
||||
if isinstance(res, (asyncio.Task, asyncio.Future)):
|
||||
if not res.done():
|
||||
@@ -858,17 +861,20 @@ def run_in_background(
|
||||
res.add_done_callback(_reset)
|
||||
return res
|
||||
|
||||
# Plain value — wrap in a resolved future
|
||||
loop = asyncio.get_event_loop()
|
||||
fut: asyncio.Future[Any] = loop.create_future()
|
||||
fut.set_result(res)
|
||||
return fut
|
||||
# Plain value — wrap in a coroutine
|
||||
async def _return_value() -> Any:
|
||||
return res
|
||||
return _return_value()
|
||||
|
||||
|
||||
def run_coroutine_in_background(
|
||||
coroutine: typing.Coroutine[Any, Any, R],
|
||||
) -> "asyncio.Task[R]":
|
||||
"""Schedule a coroutine as a background asyncio.Task."""
|
||||
) -> Any:
|
||||
"""Wrap a coroutine so it resets logcontext on completion.
|
||||
|
||||
Returns a coroutine (NOT a Task) so it's compatible with both
|
||||
Twisted's ensureDeferred and asyncio's ensure_future.
|
||||
"""
|
||||
calling_context = current_context()
|
||||
|
||||
async def _wrapper() -> R:
|
||||
@@ -877,9 +883,10 @@ def run_coroutine_in_background(
|
||||
finally:
|
||||
set_current_context(SENTINEL_CONTEXT)
|
||||
|
||||
task = asyncio.ensure_future(_wrapper())
|
||||
# Return the coroutine — let the caller decide how to schedule it
|
||||
wrapper_coro = _wrapper()
|
||||
set_current_context(calling_context)
|
||||
return task
|
||||
return wrapper_coro
|
||||
|
||||
|
||||
async def make_deferred_yieldable(awaitable: Any) -> Any:
|
||||
|
||||
+2
-2
@@ -705,8 +705,8 @@ class HomeServer(metaclass=abc.ABCMeta):
|
||||
|
||||
@cache_in_self
|
||||
def get_clock(self) -> Clock:
|
||||
# Ignore the linter error since this is the one place the `Clock` should be created.
|
||||
return Clock(self._reactor, server_name=self.hostname) # type: ignore[multiple-internal-clocks]
|
||||
from synapse.util.clock import NativeClock
|
||||
return NativeClock(self._reactor, server_name=self.hostname) # type: ignore[multiple-internal-clocks,return-value]
|
||||
|
||||
def get_datastores(self) -> Databases:
|
||||
if not self.datastores:
|
||||
|
||||
+70
-377
@@ -91,314 +91,22 @@ def make_awaitable_promise() -> Any:
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
# No running asyncio loop — use Deferred (works with Twisted reactor)
|
||||
if defer is not None:
|
||||
return defer.Deferred()
|
||||
# No running loop — use get_event_loop (set by tests/__init__.py)
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
if not loop.is_closed():
|
||||
return loop.create_future()
|
||||
except RuntimeError:
|
||||
pass
|
||||
|
||||
raise RuntimeError("Cannot create promise: no asyncio loop and no Twisted")
|
||||
raise RuntimeError("Cannot create promise: no asyncio event loop available")
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
_T = TypeVar("_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.
|
||||
|
||||
If consumeErrors is true errors will be captured from the origin deferred.
|
||||
|
||||
Cancelling or otherwise resolving an observer will not affect the original
|
||||
ObservableDeferred.
|
||||
|
||||
NB that it does not attempt to do anything with logcontexts; in general
|
||||
you should probably make_deferred_yieldable the deferreds
|
||||
returned by `observe`, and ensure that the original deferred runs its
|
||||
callbacks in the sentinel logcontext.
|
||||
"""
|
||||
|
||||
__slots__ = ["_deferred", "_observers", "_result"]
|
||||
|
||||
_deferred: "defer.Deferred[_T]"
|
||||
_observers: list["defer.Deferred[_T]"] | tuple[()]
|
||||
_result: None | tuple[Literal[True], _T] | tuple[Literal[False], Failure]
|
||||
|
||||
def __init__(self, deferred: "defer.Deferred[_T]", consumeErrors: bool = False):
|
||||
object.__setattr__(self, "_deferred", deferred)
|
||||
object.__setattr__(self, "_result", None)
|
||||
object.__setattr__(self, "_observers", [])
|
||||
|
||||
def callback(r: _T) -> _T:
|
||||
object.__setattr__(self, "_result", (True, r))
|
||||
|
||||
# once we have set _result, no more entries will be added to _observers,
|
||||
# so it's safe to replace it with the empty tuple.
|
||||
observers = self._observers
|
||||
object.__setattr__(self, "_observers", ())
|
||||
|
||||
for observer in observers:
|
||||
try:
|
||||
observer.callback(r)
|
||||
except defer.CancelledError:
|
||||
# We do not want to propagate cancellations to the original
|
||||
# deferred, or to other observers, so we can just ignore
|
||||
# this.
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.exception(
|
||||
"%r threw an exception on .callback(%r), ignoring...",
|
||||
observer,
|
||||
r,
|
||||
exc_info=e,
|
||||
)
|
||||
return r
|
||||
|
||||
def errback(f: Failure) -> Failure | None:
|
||||
object.__setattr__(self, "_result", (False, f))
|
||||
|
||||
# once we have set _result, no more entries will be added to _observers,
|
||||
# so it's safe to replace it with the empty tuple.
|
||||
observers = self._observers
|
||||
object.__setattr__(self, "_observers", ())
|
||||
|
||||
for observer in observers:
|
||||
# This is a little bit of magic to correctly propagate stack
|
||||
# traces when we `await` on one of the observer deferreds.
|
||||
f.value.__failure__ = f
|
||||
try:
|
||||
observer.errback(f)
|
||||
except defer.CancelledError:
|
||||
# We do not want to propagate cancellations to the original
|
||||
# deferred, or to other observers, so we can just ignore
|
||||
# this.
|
||||
pass
|
||||
except Exception as e:
|
||||
logger.exception(
|
||||
"%r threw an exception on .errback(%r), ignoring...",
|
||||
observer,
|
||||
f,
|
||||
exc_info=e,
|
||||
)
|
||||
|
||||
if consumeErrors:
|
||||
return None
|
||||
else:
|
||||
return f
|
||||
|
||||
deferred.addCallbacks(callback, errback)
|
||||
|
||||
@cancellable
|
||||
def observe(self) -> "defer.Deferred[_T]":
|
||||
"""Observe the underlying deferred.
|
||||
|
||||
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.
|
||||
"""
|
||||
if not self._result:
|
||||
assert isinstance(self._observers, list)
|
||||
d: "defer.Deferred[_T]" = defer.Deferred(canceller=self._remove_observer)
|
||||
self._observers.append(d)
|
||||
return d
|
||||
elif self._result[0]:
|
||||
return defer.succeed(self._result[1])
|
||||
else:
|
||||
return defer.fail(self._result[1])
|
||||
|
||||
def observers(self) -> "Collection[defer.Deferred[_T]]":
|
||||
return self._observers
|
||||
|
||||
def has_observers(self) -> bool:
|
||||
"""Returns True if there are any observers currently observing this
|
||||
ObservableDeferred.
|
||||
"""
|
||||
return bool(self._observers)
|
||||
|
||||
def has_called(self) -> bool:
|
||||
return self._result is not None
|
||||
|
||||
def has_succeeded(self) -> bool:
|
||||
return self._result is not None and self._result[0] is True
|
||||
|
||||
def get_result(self) -> _T | Failure:
|
||||
if self._result is None:
|
||||
raise ValueError(f"{self!r} has no result yet")
|
||||
return self._result[1]
|
||||
|
||||
def __getattr__(self, name: str) -> Any:
|
||||
return getattr(self._deferred, name)
|
||||
|
||||
def __setattr__(self, name: str, value: Any) -> None:
|
||||
setattr(self._deferred, name, value)
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return "<ObservableDeferred object at %s, result=%r, _deferred=%r>" % (
|
||||
id(self),
|
||||
self._result,
|
||||
self._deferred,
|
||||
)
|
||||
|
||||
def _remove_observer(self, observer: "defer.Deferred[_T]") -> None:
|
||||
"""Removes an observer from the list of observers.
|
||||
|
||||
Used as a canceller for the observer deferreds, so that if an observer
|
||||
is cancelled it is removed from the list of observers.
|
||||
"""
|
||||
if self._result is not None:
|
||||
# The underlying deferred has already resolved, so the observer has
|
||||
# already been resolved. Nothing to do.
|
||||
return
|
||||
|
||||
assert isinstance(self._observers, list)
|
||||
try:
|
||||
self._observers.remove(observer)
|
||||
except ValueError:
|
||||
# The observer was not in the list. This can happen if the underlying
|
||||
# deferred resolves at around the same time as we try to remove the
|
||||
# observer. In this case, it's possible that we tried to remove the
|
||||
# observer just after it was added to the list, but before it was
|
||||
# resolved and removed from the list by the callback/errback above.
|
||||
pass
|
||||
|
||||
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
async def concurrently_execute(
|
||||
func: Callable[[T], Any],
|
||||
args: Iterable[T],
|
||||
limit: int,
|
||||
delay_cancellation: bool = False,
|
||||
) -> None:
|
||||
"""Executes the function with each argument concurrently while limiting
|
||||
the number of concurrent executions.
|
||||
|
||||
Args:
|
||||
func: Function to execute, should return a deferred or coroutine.
|
||||
args: List of arguments to pass to func, each invocation of func
|
||||
gets a single argument.
|
||||
limit: Maximum number of conccurent executions.
|
||||
delay_cancellation: Whether to delay cancellation until after the invocations
|
||||
have finished.
|
||||
|
||||
Returns:
|
||||
None, when all function invocations have finished. The return values
|
||||
from those functions are discarded.
|
||||
"""
|
||||
it = iter(args)
|
||||
|
||||
async def _concurrently_execute_inner(value: T) -> None:
|
||||
try:
|
||||
while True:
|
||||
await maybe_awaitable(func(value))
|
||||
value = next(it)
|
||||
except StopIteration:
|
||||
pass
|
||||
|
||||
# We use `itertools.islice` to handle the case where the number of args is
|
||||
# less than the limit, avoiding needlessly spawning unnecessary background
|
||||
# tasks.
|
||||
if delay_cancellation:
|
||||
await yieldable_gather_results_delaying_cancellation(
|
||||
_concurrently_execute_inner,
|
||||
(value for value in itertools.islice(it, limit)),
|
||||
)
|
||||
else:
|
||||
await yieldable_gather_results(
|
||||
_concurrently_execute_inner,
|
||||
(value for value in itertools.islice(it, limit)),
|
||||
)
|
||||
|
||||
|
||||
P = ParamSpec("P")
|
||||
R = TypeVar("R")
|
||||
|
||||
|
||||
async def yieldable_gather_results(
|
||||
func: Callable[Concatenate[T, P], Awaitable[R]],
|
||||
iter: Iterable[T],
|
||||
*args: P.args,
|
||||
**kwargs: P.kwargs,
|
||||
) -> list[R]:
|
||||
"""Executes the function with each argument concurrently.
|
||||
|
||||
Args:
|
||||
func: Function to execute that returns an awaitable
|
||||
iter: An iterable that yields items that get passed as the first
|
||||
argument to the function
|
||||
*args: Arguments to be passed to each call to func
|
||||
**kwargs: Keyword arguments to be passed to each call to func
|
||||
|
||||
Returns
|
||||
A list containing the results of the function
|
||||
"""
|
||||
async def _run(item: T) -> R:
|
||||
return await func(item, *args, **kwargs)
|
||||
|
||||
try:
|
||||
asyncio.get_running_loop()
|
||||
results = await asyncio.gather(
|
||||
*[_run(item) for item in iter],
|
||||
return_exceptions=True,
|
||||
)
|
||||
for r in results:
|
||||
if isinstance(r, BaseException):
|
||||
raise r
|
||||
return results # type: ignore[return-value]
|
||||
except RuntimeError:
|
||||
# No asyncio loop — use Twisted fallback
|
||||
try:
|
||||
return await make_deferred_yieldable(
|
||||
defer.gatherResults(
|
||||
[run_in_background(func, item, *args, **kwargs) for item in iter],
|
||||
consumeErrors=True,
|
||||
)
|
||||
)
|
||||
except defer.FirstError as dfe:
|
||||
assert isinstance(dfe.subFailure.value, BaseException)
|
||||
raise dfe.subFailure.value from None
|
||||
|
||||
|
||||
async def yieldable_gather_results_delaying_cancellation(
|
||||
func: Callable[Concatenate[T, P], Awaitable[R]],
|
||||
iter: Iterable[T],
|
||||
*args: P.args,
|
||||
**kwargs: P.kwargs,
|
||||
) -> list[R]:
|
||||
"""Executes the function with each argument concurrently.
|
||||
Cancellation is delayed until after all the results have been gathered.
|
||||
|
||||
See `yieldable_gather_results`.
|
||||
|
||||
Args:
|
||||
func: Function to execute that returns an awaitable
|
||||
iter: An iterable that yields items that get passed as the first
|
||||
argument to the function
|
||||
*args: Arguments to be passed to each call to func
|
||||
**kwargs: Keyword arguments to be passed to each call to func
|
||||
|
||||
Returns
|
||||
A list containing the results of the function
|
||||
"""
|
||||
# Use asyncio.shield to delay cancellation
|
||||
async def _run(item: T) -> R:
|
||||
return await func(item, *args, **kwargs)
|
||||
|
||||
results = await asyncio.shield(
|
||||
asyncio.gather(
|
||||
*[_run(item) for item in iter],
|
||||
return_exceptions=True,
|
||||
)
|
||||
)
|
||||
|
||||
for r in results:
|
||||
if isinstance(r, BaseException):
|
||||
raise r
|
||||
|
||||
return results # type: ignore[return-value]
|
||||
|
||||
|
||||
P = ParamSpec("P")
|
||||
T1 = TypeVar("T1")
|
||||
T2 = TypeVar("T2")
|
||||
T3 = TypeVar("T3")
|
||||
@@ -407,60 +115,62 @@ T5 = TypeVar("T5")
|
||||
T6 = TypeVar("T6")
|
||||
|
||||
|
||||
@overload
|
||||
def gather_results(
|
||||
deferredList: tuple[()], consumeErrors: bool = ...
|
||||
) -> "defer.Deferred[tuple[()]]": ...
|
||||
|
||||
|
||||
@overload
|
||||
def gather_results(
|
||||
deferredList: tuple["defer.Deferred[T1]"],
|
||||
consumeErrors: bool = ...,
|
||||
) -> "defer.Deferred[tuple[T1]]": ...
|
||||
|
||||
|
||||
@overload
|
||||
def gather_results(
|
||||
deferredList: tuple["defer.Deferred[T1]", "defer.Deferred[T2]"],
|
||||
consumeErrors: bool = ...,
|
||||
) -> "defer.Deferred[tuple[T1, T2]]": ...
|
||||
|
||||
|
||||
@overload
|
||||
def gather_results(
|
||||
deferredList: tuple[
|
||||
"defer.Deferred[T1]", "defer.Deferred[T2]", "defer.Deferred[T3]"
|
||||
],
|
||||
consumeErrors: bool = ...,
|
||||
) -> "defer.Deferred[tuple[T1, T2, T3]]": ...
|
||||
|
||||
|
||||
@overload
|
||||
def gather_results(
|
||||
deferredList: tuple[
|
||||
"defer.Deferred[T1]",
|
||||
"defer.Deferred[T2]",
|
||||
"defer.Deferred[T3]",
|
||||
"defer.Deferred[T4]",
|
||||
],
|
||||
consumeErrors: bool = ...,
|
||||
) -> "defer.Deferred[tuple[T1, T2, T3, T4]]": ...
|
||||
|
||||
|
||||
def gather_results( # type: ignore[misc]
|
||||
deferredList: tuple["defer.Deferred[T1]", ...],
|
||||
async def gather_results(
|
||||
awaitableList: tuple[Any, ...],
|
||||
consumeErrors: bool = False,
|
||||
) -> "defer.Deferred[tuple[T1, ...]]":
|
||||
"""Combines a tuple of `Deferred`s into a single `Deferred`.
|
||||
) -> tuple[Any, ...]:
|
||||
"""Gather multiple awaitables into a single tuple of results.
|
||||
|
||||
Wraps `defer.gatherResults` to provide type annotations that support heterogenous
|
||||
lists of `Deferred`s.
|
||||
asyncio-native replacement for the old Deferred-based gather_results.
|
||||
"""
|
||||
# The `type: ignore[misc]` above suppresses
|
||||
# "Overloaded function implementation cannot produce return type of signature 1/2/3"
|
||||
deferred = defer.gatherResults(deferredList, consumeErrors=consumeErrors)
|
||||
return deferred.addCallback(tuple)
|
||||
results = await asyncio.gather(*awaitableList, return_exceptions=consumeErrors)
|
||||
if not consumeErrors:
|
||||
for r in results:
|
||||
if isinstance(r, BaseException):
|
||||
raise r
|
||||
return tuple(results)
|
||||
|
||||
|
||||
async def yieldable_gather_results(
|
||||
func: Callable[..., Awaitable[R]],
|
||||
iter: Iterable[Any],
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> list[R]:
|
||||
"""Execute func with each item concurrently using asyncio.gather."""
|
||||
async def _run(item: Any) -> R:
|
||||
return await func(item, *args, **kwargs)
|
||||
|
||||
results = await asyncio.gather(
|
||||
*[_run(item) for item in iter],
|
||||
return_exceptions=True,
|
||||
)
|
||||
for r in results:
|
||||
if isinstance(r, BaseException):
|
||||
raise r
|
||||
return list(results)
|
||||
|
||||
|
||||
async def yieldable_gather_results_delaying_cancellation(
|
||||
func: Callable[..., Awaitable[R]],
|
||||
iter: Iterable[Any],
|
||||
*args: Any,
|
||||
**kwargs: Any,
|
||||
) -> list[R]:
|
||||
"""Like yieldable_gather_results but shielded from cancellation."""
|
||||
async def _run(item: Any) -> R:
|
||||
return await func(item, *args, **kwargs)
|
||||
|
||||
results = await asyncio.shield(
|
||||
asyncio.gather(
|
||||
*[_run(item) for item in iter],
|
||||
return_exceptions=True,
|
||||
)
|
||||
)
|
||||
for r in results:
|
||||
if isinstance(r, BaseException):
|
||||
raise r
|
||||
return list(results)
|
||||
|
||||
|
||||
@overload
|
||||
@@ -550,31 +260,12 @@ async def gather_optional_coroutines(
|
||||
overload above.
|
||||
"""
|
||||
|
||||
# Use asyncio.gather if an event loop is running, otherwise fall back to
|
||||
# Twisted's defer.gatherResults during transition
|
||||
try:
|
||||
asyncio.get_running_loop()
|
||||
tasks = [
|
||||
asyncio.ensure_future(coroutine)
|
||||
for coroutine in coroutines
|
||||
if coroutine is not None
|
||||
]
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
for r in results:
|
||||
if isinstance(r, BaseException):
|
||||
raise r
|
||||
except RuntimeError:
|
||||
# No asyncio loop — use Twisted
|
||||
results = await make_deferred_yieldable(
|
||||
defer.gatherResults(
|
||||
[
|
||||
run_coroutine_in_background(coroutine)
|
||||
for coroutine in coroutines
|
||||
if coroutine is not None
|
||||
],
|
||||
consumeErrors=True,
|
||||
)
|
||||
)
|
||||
tasks = [coroutine for coroutine in coroutines if coroutine is not None]
|
||||
results = await asyncio.gather(*tasks, return_exceptions=True)
|
||||
|
||||
for r in results:
|
||||
if isinstance(r, BaseException):
|
||||
raise r
|
||||
|
||||
results_iter = iter(results)
|
||||
return tuple(
|
||||
@@ -1013,3 +704,5 @@ Linearizer = NativeLinearizer # type: ignore[misc]
|
||||
ReadWriteLock = NativeReadWriteLock # type: ignore[misc]
|
||||
AwakenableSleeper = NativeAwakenableSleeper # type: ignore[misc]
|
||||
DeferredEvent = NativeEvent # type: ignore[misc]
|
||||
ObservableDeferred = ObservableFuture # type: ignore[misc]
|
||||
concurrently_execute = native_concurrently_execute # type: ignore[misc]
|
||||
|
||||
+46
-8
@@ -710,7 +710,8 @@ class NativeClock:
|
||||
server_name: The server name for logging context.
|
||||
"""
|
||||
|
||||
def __init__(self, server_name: str) -> None:
|
||||
def __init__(self, reactor: Any = None, server_name: str = "") -> None:
|
||||
# reactor arg accepted for backward compatibility but ignored
|
||||
self._server_name = server_name
|
||||
self._delayed_call_id: int = 0
|
||||
self._looping_calls: WeakSet[NativeLoopingCall] = WeakSet()
|
||||
@@ -720,9 +721,19 @@ class NativeClock:
|
||||
# Lazily initialized when first needed
|
||||
self._loop: asyncio.AbstractEventLoop | None = None
|
||||
|
||||
# Internal timer system for fake time support.
|
||||
# Pending sleeps: list of (wake_time, future)
|
||||
import heapq
|
||||
self._fake_time: float = time_mod.time()
|
||||
self._pending_sleeps: list[tuple[float, asyncio.Future]] = []
|
||||
self._use_fake_time = False # Set to True by tests
|
||||
|
||||
def _get_loop(self) -> asyncio.AbstractEventLoop:
|
||||
if self._loop is None:
|
||||
self._loop = asyncio.get_running_loop()
|
||||
try:
|
||||
self._loop = asyncio.get_running_loop()
|
||||
except RuntimeError:
|
||||
self._loop = asyncio.get_event_loop()
|
||||
return self._loop
|
||||
|
||||
def shutdown(self) -> None:
|
||||
@@ -731,15 +742,41 @@ class NativeClock:
|
||||
self.cancel_all_delayed_calls()
|
||||
|
||||
def time(self) -> float:
|
||||
"""Returns the current system time in seconds since epoch."""
|
||||
"""Returns the current time in seconds since epoch."""
|
||||
if self._use_fake_time:
|
||||
return self._fake_time
|
||||
return time_mod.time()
|
||||
|
||||
def time_msec(self) -> int:
|
||||
"""Returns the current system time in milliseconds since epoch."""
|
||||
"""Returns the current time in milliseconds since epoch."""
|
||||
return int(self.time() * 1000)
|
||||
|
||||
async def sleep(self, duration: Duration) -> None:
|
||||
await asyncio.sleep(duration.as_secs())
|
||||
"""Sleep for duration, using fake time if enabled."""
|
||||
if self._use_fake_time:
|
||||
import heapq
|
||||
loop = self._get_loop()
|
||||
future: asyncio.Future[None] = loop.create_future()
|
||||
wake_time = self._fake_time + duration.as_secs()
|
||||
heapq.heappush(self._pending_sleeps, (wake_time, future))
|
||||
await future
|
||||
else:
|
||||
await asyncio.sleep(duration.as_secs())
|
||||
|
||||
def advance(self, seconds: float) -> None:
|
||||
"""Advance fake time by seconds, firing any due sleeps.
|
||||
|
||||
Used by tests to control time deterministically.
|
||||
"""
|
||||
self._use_fake_time = True
|
||||
self._fake_time += seconds
|
||||
|
||||
# Fire any sleeps that are now due
|
||||
while self._pending_sleeps and self._pending_sleeps[0][0] <= self._fake_time:
|
||||
import heapq
|
||||
_, future = heapq.heappop(self._pending_sleeps)
|
||||
if not future.done():
|
||||
future.set_result(None)
|
||||
|
||||
def looping_call(
|
||||
self,
|
||||
@@ -777,7 +814,7 @@ class NativeClock:
|
||||
|
||||
async def _loop() -> None:
|
||||
if not now:
|
||||
await asyncio.sleep(interval)
|
||||
await self.sleep(Duration(seconds=interval))
|
||||
|
||||
while True:
|
||||
try:
|
||||
@@ -797,9 +834,10 @@ class NativeClock:
|
||||
except Exception:
|
||||
logger.exception("Looping call %s died", instance_id)
|
||||
|
||||
await asyncio.sleep(interval)
|
||||
await self.sleep(Duration(seconds=interval))
|
||||
|
||||
task_obj = asyncio.create_task(_loop())
|
||||
loop = asyncio.get_event_loop()
|
||||
task_obj = loop.create_task(_loop())
|
||||
call = NativeLoopingCall(task_obj)
|
||||
self._looping_calls.add(call)
|
||||
|
||||
|
||||
+12
-3
@@ -307,14 +307,22 @@ class FakeChannel:
|
||||
"""
|
||||
Wait until the request is finished.
|
||||
"""
|
||||
end_time = self._reactor.seconds() + timeout_ms / 1000.0
|
||||
import asyncio
|
||||
import time as _time
|
||||
|
||||
deadline = _time.monotonic() + timeout_ms / 1000.0
|
||||
self._reactor.run()
|
||||
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
while not self.is_finished():
|
||||
if self._reactor.seconds() > end_time:
|
||||
if _time.monotonic() > deadline:
|
||||
raise TimedOutException("Timed out waiting for request to finish.")
|
||||
|
||||
self._reactor.advance(0.1)
|
||||
# Drive asyncio event loop for DB operations, task completions, etc.
|
||||
if not loop.is_closed() and not loop.is_running():
|
||||
loop.run_until_complete(asyncio.sleep(0))
|
||||
|
||||
def extract_cookies(self, cookies: MutableMapping[str, str]) -> None:
|
||||
"""Process the contents of any Set-Cookie headers in the response
|
||||
@@ -833,7 +841,8 @@ def get_clock() -> tuple[ThreadedMemoryReactorClock, Clock]:
|
||||
# Ignore the linter error since this is an expected usage of creating a `Clock` for
|
||||
# testing purposes.
|
||||
reactor = ThreadedMemoryReactorClock()
|
||||
hs_clock = Clock(reactor, server_name="test_server") # type: ignore[multiple-internal-clocks]
|
||||
from synapse.util.clock import NativeClock
|
||||
hs_clock = NativeClock(reactor, server_name="test_server") # type: ignore[multiple-internal-clocks]
|
||||
return reactor, hs_clock
|
||||
|
||||
|
||||
|
||||
+17
-13
@@ -869,30 +869,34 @@ class HomeserverTestCase(TestCase):
|
||||
|
||||
def pump(self, by: float = 0.0) -> None:
|
||||
"""
|
||||
Pump both the test reactor and the asyncio event loop.
|
||||
Pump both the test reactor and the asyncio event loop,
|
||||
advancing fake time on the NativeClock.
|
||||
"""
|
||||
import asyncio
|
||||
|
||||
# Advance Twisted's fake clock
|
||||
self.reactor.pump([by] * 100)
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
# Drive the asyncio event loop
|
||||
try:
|
||||
loop = asyncio.get_event_loop()
|
||||
if not loop.is_closed() and not loop.is_running():
|
||||
loop.run_until_complete(asyncio.sleep(0))
|
||||
except RuntimeError:
|
||||
pass
|
||||
# Advance fake time on the clock (fires pending sleeps)
|
||||
self.clock.advance(by)
|
||||
|
||||
# Advance Twisted's fake clock too (for any Twisted-driven code)
|
||||
self.reactor.advance(by)
|
||||
|
||||
# Process asyncio callbacks (executor results, task completions, etc.)
|
||||
if not loop.is_closed() and not loop.is_running():
|
||||
loop.run_until_complete(asyncio.sleep(0))
|
||||
|
||||
def get_success(self, d: Awaitable[TV], by: float = 0.0) -> TV:
|
||||
import asyncio
|
||||
|
||||
# Advance Twisted's fake clock first (for any time-dependent setup)
|
||||
loop = asyncio.get_event_loop()
|
||||
|
||||
# Pump the fake reactor first if time advancement is needed
|
||||
if by > 0:
|
||||
self.reactor.pump([by] * 100)
|
||||
|
||||
# Drive the coroutine to completion on the global event loop
|
||||
loop = asyncio.get_event_loop()
|
||||
# Run the awaitable to completion on the asyncio loop.
|
||||
# nest_asyncio allows this even if the loop is already running.
|
||||
return loop.run_until_complete(d) # type: ignore[arg-type]
|
||||
|
||||
def get_failure(
|
||||
|
||||
Reference in New Issue
Block a user