dandri/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse.git synced 2026-07-16 18:51:58 +00:00
Code Issues Packages Projects Releases Wiki Activity

⏺ The remaining 37 break down as:

Browse Source 
- 11 http/client.py — Twisted HTTP client (needs full aiohttp rewrite)
  - 5 connectproxyclient.py — Twisted proxy client (needs full rewrite)
  - 4 background_process_metrics.py — type annotations (harmless)
  - 4 _scripts/synapse_port_db.py — CLI script (low priority)
  - 3 server.py — startup/shutdown (type annotations + defer.ensureDeferred)
  - 3 app/_base.py — app startup (defer.ensureDeferred)
  - 2 http/server.py — type annotation + isinstance check
  - 1 manhole.py — interactive manhole
  - 1 util/__init__.py — unwrapFirstError compat
  - 1 crypto/keyring.py — type annotation
  - 1 admin_cmd.py — CLI script
  - 1 _scripts/update_synapse_database.py — CLI script

  Here's a summary of what was accomplished in this session:

  Completed: Replaced all .addErrback/.addCallback/.addBoth calls on production code that was receiving asyncio Tasks instead of Deferreds. Files fixed:

  1. cached_call.py — Replaced Deferred/Failure/.addBoth with asyncio.Task/add_done_callback
  2. response_cache.py — Replaced .addBoth with add_done_callback, .called with has_called()
  3. push_rule.py — Replaced gather_results(...).addErrback(unwrapFirstError) with await gather_results(...)
  4. bulk_push_rule_evaluator.py — Same pattern
  5. events_worker.py — Replaced defer.Deferred() with asyncio.Future(), defer.gatherResults with asyncio.gather, .errback()/.callback() with .set_exception()/.set_result()
  6. partial_state_events_tracker.py — Same pattern, both PartialStateEventsTracker and PartialCurrentStateTracker
  7. initial_sync.py — Replaced .addCallback chain with async wrapper, removed gather_results().addErrback()
  8. message.py — Replaced .addErrback on Task with try/except wrapper
  9. federation_server.py — Removed gather_results().addErrback()
  10. transactions.py — Replaced .addErrback with add_done_callback
  11. e2e_keys.py, appservice.py, sender/__init__.py, stream.py, persist_events.py, notifier.py, batching_queue.py — All migrated from Twisted to asyncio

  Also fixed:
  - NativeLoopingCall missing .running property
  - Heapq tiebreaker for _pending_sleeps (Future comparison error)
  - FakeChannel.await_result now advances NativeClock fake time
  - run_in_background returns resolved Futures for plain values (not unawaited coroutines)

  Remaining 37 defer.* calls are in HTTP client (needs full rewrite), type annotations, startup code, and CLI scripts — all expected.
This commit is contained in:
Matthew Hodgson
2026-03-22 02:20:59 +00:00
parent a499d536b0
commit ac8077bcee
22 changed files with 341 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files
synapse/federation/federation_server.py
+6 -11
View File
@@ -69,7 +69,6 @@ from synapse.federation.units import Edu, Transaction, serialize_and_filter_pdus
from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME
from synapse.http.servlet import assert_params_in_dict
from synapse.logging.context import (
make_deferred_yieldable,
nested_logging_context,
run_in_background,
)
@@ -90,7 +89,6 @@ from synapse.storage.databases.main.lock import Lock
from synapse.storage.databases.main.roommember import extract_heroes_from_room_summary
from synapse.storage.roommember import MemberSummary
from synapse.types import JsonDict, StateMap, UserID, get_domain_from_id
from synapse.util import unwrapFirstError
from synapse.util.async_helpers import Linearizer, concurrently_execute, gather_results
from synapse.util.caches.response_cache import ResponseCache
from synapse.util.duration import Duration
@@ -408,16 +406,13 @@ class FederationServer(FederationBase):
# We process PDUs and EDUs in parallel. This is important as we don't
# want to block things like to device messages from reaching clients
# behind the potentially expensive handling of PDUs.
pdu_results, _ = await make_deferred_yieldable(
gather_results(
(
run_in_background(
self._handle_pdus_in_txn, origin, transaction, request_time
),
run_in_background(self._handle_edus_in_txn, origin, transaction),
pdu_results, _ = await gather_results(
(
run_in_background(
self._handle_pdus_in_txn, origin, transaction, request_time
),
consumeErrors=True,
).addErrback(unwrapFirstError)
run_in_background(self._handle_edus_in_txn, origin, transaction),
),
)
response = {"pdus": pdu_results}
synapse/federation/sender/__init__.py
+7 -13
View File
@@ -143,10 +143,7 @@ from typing import (
import attr
from prometheus_client import Counter
try:
from twisted.internet import defer
except ImportError:
pass
import asyncio
import synapse.metrics
from synapse.api.constants import EventTypes, Membership
@@ -158,7 +155,7 @@ from synapse.federation.sender.per_destination_queue import (
)
from synapse.federation.sender.transaction_manager import TransactionManager
from synapse.federation.units import Edu
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.context import run_in_background
from synapse.metrics import (
SERVER_NAME_LABEL,
LaterGauge,
@@ -746,14 +743,11 @@ class FederationSender(AbstractFederationSender):
event = event_cache.event
events_by_room.setdefault(event.room_id, []).append(event)
await make_deferred_yieldable(
defer.gatherResults(
[
run_in_background(handle_room_events, evs)
for evs in events_by_room.values()
],
consumeErrors=True,
)
await asyncio.gather(
*(
run_in_background(handle_room_events, evs)
for evs in events_by_room.values()
),
)
logger.debug("Successfully handled up to %i", next_token)
synapse/handlers/appservice.py
+43 -55
View File
@@ -28,17 +28,14 @@ from typing import (
from prometheus_client import Counter
try:
from twisted.internet import defer
except ImportError:
pass
import asyncio
import synapse
from synapse.api.constants import EduTypes, EventTypes
from synapse.appservice import ApplicationService
from synapse.events import EventBase
from synapse.handlers.presence import format_user_presence_state
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.context import run_in_background
from synapse.metrics import (
SERVER_NAME_LABEL,
event_processing_loop_counter,
@@ -191,14 +188,11 @@ class ApplicationServicesHandler:
for event in events:
await handle_event(event)
await make_deferred_yieldable(
defer.gatherResults(
[
run_in_background(handle_room_events, evs)
for evs in events_by_room.values()
],
consumeErrors=True,
)
await asyncio.gather(
*(
run_in_background(handle_room_events, evs)
for evs in events_by_room.values()
),
)
await self.store.set_appservice_last_pos(upper_bound)
@@ -763,21 +757,19 @@ class ApplicationServicesHandler:
) -> list[JsonDict]:
services = self._get_services_for_3pn(protocol)
results = await make_deferred_yieldable(
defer.DeferredList(
[
run_in_background(
self.appservice_api.query_3pe, service, kind, protocol, fields
)
for service in services
],
consumeErrors=True,
)
results = await asyncio.gather(
*(
run_in_background(
self.appservice_api.query_3pe, service, kind, protocol, fields
)
for service in services
),
return_exceptions=True,
)
ret = []
for success, result in results:
if success:
for result in results:
if not isinstance(result, BaseException):
ret.extend(result)
return ret
@@ -910,26 +902,24 @@ class ApplicationServicesHandler:
continue
# Query each service in parallel.
results = await make_deferred_yieldable(
defer.DeferredList(
[
run_in_background( # type: ignore[call-overload]
self.appservice_api.claim_client_keys,
# We know this must be an app service.
self.store.get_app_service_by_id(service_id),
service_query,
)
for service_id, service_query in query_by_appservice.items()
],
consumeErrors=True,
)
results = await asyncio.gather(
*(
run_in_background( # type: ignore[call-overload]
self.appservice_api.claim_client_keys,
# We know this must be an app service.
self.store.get_app_service_by_id(service_id),
service_query,
)
for service_id, service_query in query_by_appservice.items()
),
return_exceptions=True,
)
# Patch together the results -- they are all independent (since they
# require exclusive control over the users, which is the outermost key).
claimed_keys: dict[str, dict[str, dict[str, JsonDict]]] = {}
for success, result in results:
if success:
for result in results:
if not isinstance(result, BaseException):
claimed_keys.update(result[0])
missing.extend(result[1])
@@ -966,27 +956,25 @@ class ApplicationServicesHandler:
continue
# Query each service in parallel.
results = await make_deferred_yieldable(
defer.DeferredList(
[
run_in_background( # type: ignore[call-overload]
self.appservice_api.query_keys,
# We know this must be an app service.
self.store.get_app_service_by_id(service_id),
service_query,
)
for service_id, service_query in query_by_appservice.items()
],
consumeErrors=True,
)
results = await asyncio.gather(
*(
run_in_background( # type: ignore[call-overload]
self.appservice_api.query_keys,
# We know this must be an app service.
self.store.get_app_service_by_id(service_id),
service_query,
)
for service_id, service_query in query_by_appservice.items()
),
return_exceptions=True,
)
# Patch together the results -- they are all independent (since they
# require exclusive control over the users). They get returned as a single
# dictionary.
key_queries: dict[str, dict[str, dict[str, JsonDict]]] = {}
for success, result in results:
if success:
for result in results:
if not isinstance(result, BaseException):
key_queries.update(result)
return key_queries
synapse/handlers/e2e_keys.py
+7 -13
View File
@@ -28,15 +28,12 @@ from signedjson.key import VerifyKey, decode_verify_key_bytes
from signedjson.sign import SignatureVerifyException, verify_signed_json
from unpaddedbase64 import decode_base64
try:
from twisted.internet import defer
except ImportError:
pass
import asyncio
from synapse.api.constants import EduTypes
from synapse.api.errors import CodeMessageException, Codes, NotFoundError, SynapseError
from synapse.handlers.device import DeviceWriterHandler
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.context import run_in_background
from synapse.logging.opentracing import log_kv, set_tag, tag_args, trace
from synapse.types import (
JsonDict,
@@ -809,14 +806,11 @@ class E2eKeysHandler:
set_tag("error", True)
set_tag("reason", str(failure))
await make_deferred_yieldable(
defer.gatherResults(
[
run_in_background(claim_client_keys, destination)
for destination in remote_queries
],
consumeErrors=True,
)
await asyncio.gather(
*(
run_in_background(claim_client_keys, destination)
for destination in remote_queries
),
)
logger.info(
synapse/handlers/initial_sync.py
+33 -33
View File
@@ -34,7 +34,7 @@ from synapse.events.utils import SerializeEventConfig
from synapse.events.validator import EventValidator
from synapse.handlers.presence import format_user_presence_state
from synapse.handlers.receipts import ReceiptEventSource
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.context import run_in_background
from synapse.storage.roommember import RoomsForUser
from synapse.streams.config import PaginationConfig
from synapse.types import (
@@ -46,7 +46,6 @@ from synapse.types import (
StreamToken,
UserID,
)
from synapse.util import unwrapFirstError
from synapse.util.async_helpers import concurrently_execute, gather_results
from synapse.util.caches.response_cache import ResponseCache
from synapse.visibility import filter_and_transform_events_for_client
@@ -199,31 +198,35 @@ class InitialSyncHandler:
try:
if event.membership == Membership.JOIN:
room_end_token = now_token.room_key
deferred_room_state = run_in_background(
self._state_storage_controller.get_current_state, event.room_id
)
async def _get_current_state() -> dict:
return await self._state_storage_controller.get_current_state(event.room_id)
deferred_room_state = run_in_background(_get_current_state)
elif event.membership == Membership.LEAVE:
room_end_token = RoomStreamToken(
stream=event.event_pos.stream,
)
deferred_room_state = run_in_background(
self._state_storage_controller.get_state_for_events,
[event.event_id],
).addCallback(lambda states: states[event.event_id])
(messages, token), current_state = await make_deferred_yieldable(
gather_results(
(
run_in_background(
self.store.get_recent_events_for_room,
event.room_id,
limit=pagin_config.limit,
end_token=room_end_token,
),
deferred_room_state,
async def _get_state_for_event() -> dict:
states = await self._state_storage_controller.get_state_for_events(
[event.event_id],
)
return states[event.event_id]
deferred_room_state = run_in_background(_get_state_for_event)
(messages, token), current_state = await gather_results(
(
run_in_background(
self.store.get_recent_events_for_room,
event.room_id,
limit=pagin_config.limit,
end_token=room_end_token,
),
deferred_room_state,
)
).addErrback(unwrapFirstError)
)
messages = await filter_and_transform_events_for_client(
self._storage_controllers,
@@ -480,20 +483,17 @@ class InitialSyncHandler:
requester.user.to_string(),
)
presence, receipts, (messages, token) = await make_deferred_yieldable(
gather_results(
(
run_in_background(get_presence),
run_in_background(get_receipts),
run_in_background(
self.store.get_recent_events_for_room,
room_id,
limit=pagin_config.limit,
end_token=now_token.room_key,
),
presence, receipts, (messages, token) = await gather_results(
(
run_in_background(get_presence),
run_in_background(get_receipts),
run_in_background(
self.store.get_recent_events_for_room,
room_id,
limit=pagin_config.limit,
end_token=now_token.room_key,
),
consumeErrors=True,
).addErrback(unwrapFirstError)
),
)
messages = await filter_and_transform_events_for_client(
synapse/handlers/message.py
+12 -12
View File
@@ -69,7 +69,7 @@ from synapse.events.validator import EventValidator
from synapse.handlers.directory import DirectoryHandler
from synapse.handlers.worker_lock import NEW_EVENT_DURING_PURGE_LOCK_NAME
from synapse.logging import opentracing
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.context import run_in_background
from synapse.replication.http.send_events import ReplicationSendEventsRestServlet
from synapse.storage.databases.main.events_worker import EventRedactBehaviour
from synapse.types import (
@@ -83,7 +83,7 @@ from synapse.types import (
create_requester,
)
from synapse.types.state import StateFilter
from synapse.util import log_failure, unwrapFirstError
from synapse.util import log_failure
from synapse.util.async_helpers import Linearizer, gather_results
from synapse.util.caches.expiringcache import ExpiringCache
from synapse.util.duration import Duration
@@ -1579,10 +1579,14 @@ class EventCreationHandler:
# We now persist the event (and update the cache in parallel, since we
# don't want to block on it).
#
# Note: mypy gets confused if we inline dl and check with twisted#11770.
# Some kind of bug in mypy's deduction?
deferreds = (
async def _cache_joined_hosts() -> None:
try:
await self.cache_joined_hosts_for_events(events_and_context)
except Exception:
log_failure(None, "cache_joined_hosts_for_event failed")
tasks = (
run_in_background(
self._persist_events,
requester=requester,
@@ -1590,13 +1594,9 @@ class EventCreationHandler:
ratelimit=ratelimit,
extra_users=extra_users,
),
run_in_background(
self.cache_joined_hosts_for_events, events_and_context
).addErrback(log_failure, "cache_joined_hosts_for_event failed"),
run_in_background(_cache_joined_hosts),
)
result, _ = await make_deferred_yieldable(
gather_results(deferreds, consumeErrors=True)
).addErrback(unwrapFirstError)
result, _ = await gather_results(tasks)
return result
synapse/logging/context.py
+10 -8
View File
@@ -841,10 +841,11 @@ def run_in_background(
try:
res = f(*args, **kwargs)
except Exception as e:
# Return a coroutine that raises the exception
async def _raise() -> Any:
raise e
return _raise()
# Return a failed Task so callers can handle it asynchronously
loop = asyncio.get_event_loop()
future: asyncio.Future[Any] = loop.create_future()
future.set_exception(e)
return future
if isinstance(res, typing.Coroutine):
# Schedule the coroutine as a Task on the event loop.
@@ -861,10 +862,11 @@ def run_in_background(
res.add_done_callback(_reset)
return res
# Plain value — wrap in a coroutine
async def _return_value() -> Any:
return res
return _return_value()
# Plain value — return a resolved Future so callers can await it
loop = asyncio.get_event_loop()
future: asyncio.Future[Any] = loop.create_future()
future.set_result(res)
return future
def run_coroutine_in_background(
synapse/notifier.py
+27 -21
View File
@@ -35,11 +35,7 @@ from typing import (
import attr
from prometheus_client import Counter
try:
from twisted.internet import defer
from twisted.internet.defer import Deferred
except ImportError:
pass
import asyncio
from synapse.api.constants import EduTypes, EventTypes, HistoryVisibility, Membership
from synapse.api.errors import AuthError
@@ -147,13 +143,13 @@ class _NotifierUserStream:
self.last_notified_ms = time_now_ms
# Set of listeners that we need to wake up when there has been a change.
self.listeners: set[Deferred[StreamToken]] = set()
self.listeners: set[asyncio.Future[StreamToken]] = set()
def update_and_fetch_deferreds(
self,
current_token: StreamToken,
time_now_ms: int,
) -> Collection["Deferred[StreamToken]"]:
) -> Collection["asyncio.Future[StreamToken]"]:
"""Update the stream for this user because of a new event from an
event source, and return the set of deferreds to wake up.
@@ -189,27 +185,35 @@ class _NotifierUserStream:
def count_listeners(self) -> int:
return len(self.listeners)
def new_listener(self, token: StreamToken) -> "Deferred[StreamToken]":
"""Returns a deferred that is resolved when there is a new token
def new_listener(self, token: StreamToken) -> "asyncio.Future[StreamToken]":
"""Returns a future that is resolved when there is a new token
greater than the given token.
Args:
token: The token from which we are streaming from, i.e. we shouldn't
notify for things that happened before this.
"""
loop = asyncio.get_running_loop()
# Immediately wake up stream if something has already since happened
# since their last token.
if token != self.current_token:
return defer.succeed(self.current_token)
f: asyncio.Future[StreamToken] = loop.create_future()
f.set_result(self.current_token)
return f
# Create a new deferred and add it to the set of listeners. We add a
# Create a new future and add it to the set of listeners. We add a
# cancel handler to remove it from the set again, to handle timeouts.
deferred: "Deferred[StreamToken]" = Deferred(
canceller=lambda d: self.listeners.discard(d)
)
self.listeners.add(deferred)
future: asyncio.Future[StreamToken] = loop.create_future()
self.listeners.add(future)
return deferred
def _on_cancel(f: asyncio.Future[StreamToken]) -> None:
if f.cancelled():
self.listeners.discard(f)
future.add_done_callback(_on_cancel)
return future
@attr.s(slots=True, frozen=True, auto_attribs=True)
@@ -370,7 +374,7 @@ class Notifier:
time_now_ms = self.clock.time_msec()
current_token = self.event_sources.get_current_token()
listeners: list["Deferred[StreamToken]"] = []
listeners: list["asyncio.Future[StreamToken]"] = []
for user_stream in user_streams:
try:
listeners.extend(
@@ -381,7 +385,8 @@ class Notifier:
with PreserveLoggingContext():
for listener in listeners:
listener.callback(current_token)
if not listener.done():
listener.set_result(current_token)
users_woken_by_stream_counter.labels(
stream=StreamKeyType.UN_PARTIAL_STATED_ROOMS,
@@ -591,7 +596,7 @@ class Notifier:
time_now_ms = self.clock.time_msec()
current_token = self.event_sources.get_current_token()
listeners: list["Deferred[StreamToken]"] = []
listeners: list["asyncio.Future[StreamToken]"] = []
for user_stream in user_streams:
try:
listeners.extend(
@@ -602,13 +607,14 @@ class Notifier:
except Exception:
logger.exception("Failed to notify listener")
# We resolve all these deferreds in one go so that we only need to
# We resolve all these futures in one go so that we only need to
# call `PreserveLoggingContext` once, as it has a bunch of overhead
# (to calculate performance stats)
if listeners:
with PreserveLoggingContext():
for listener in listeners:
listener.callback(current_token)
if not listener.done():
listener.set_result(current_token)
if user_streams:
users_woken_by_stream_counter.labels(
synapse/push/bulk_push_rule_evaluator.py
+20 -33
View File
@@ -26,16 +26,10 @@ from typing import (
Collection,
Mapping,
Sequence,
cast,
)
from prometheus_client import Counter
try:
from twisted.internet.defer import Deferred
except ImportError:
pass
from synapse.api.constants import (
MAIN_TIMELINE,
EventContentFields,
@@ -47,7 +41,7 @@ from synapse.api.room_versions import PushRuleRoomFlag
from synapse.event_auth import auth_types_for_event, get_user_power_level
from synapse.events import EventBase, relation_from_event
from synapse.events.snapshot import EventContext, EventPersistencePair
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.context import run_in_background
from synapse.metrics import SERVER_NAME_LABEL
from synapse.state import CREATE_KEY, POWER_KEY
from synapse.storage.databases.main.roommember import EventIdMembership
@@ -56,7 +50,6 @@ from synapse.storage.roommember import ProfileInfo
from synapse.synapse_rust.push import FilteredPushRules, PushRuleEvaluator
from synapse.types import JsonValue
from synapse.types.state import StateFilter
from synapse.util import unwrapFirstError
from synapse.util.async_helpers import gather_results
from synapse.util.caches import register_cache
from synapse.util.metrics import measure_func
@@ -404,31 +397,25 @@ class BulkPushRuleEvaluator:
(power_levels, sender_power_level),
related_events,
profiles,
) = await make_deferred_yieldable(
cast(
"Deferred[tuple[int, tuple[dict, int | None], dict[str, dict[str, JsonValue]], Mapping[str, ProfileInfo]]]",
gather_results(
(
run_in_background( # type: ignore[call-overload]
self.store.get_number_joined_users_in_room,
event.room_id, # type: ignore[arg-type]
),
run_in_background(
self._get_power_levels_and_sender_level,
event,
context,
event_id_to_event,
),
run_in_background(self._related_events, event),
run_in_background( # type: ignore[call-overload]
self.store.get_subset_users_in_room_with_profiles,
event.room_id,
rules_by_user.keys(),
),
),
consumeErrors=True,
).addErrback(unwrapFirstError),
)
) = await gather_results(
(
run_in_background( # type: ignore[call-overload]
self.store.get_number_joined_users_in_room,
event.room_id, # type: ignore[arg-type]
),
run_in_background(
self._get_power_levels_and_sender_level,
event,
context,
event_id_to_event,
),
run_in_background(self._related_events, event),
run_in_background( # type: ignore[call-overload]
self.store.get_subset_users_in_room_with_profiles,
event.room_id,
rules_by_user.keys(),
),
),
)
# Find the event's thread ID.
synapse/rest/client/transactions.py
+6 -7
View File
@@ -27,9 +27,9 @@ from typing import TYPE_CHECKING, Awaitable, Callable, Hashable
from typing_extensions import ParamSpec
import asyncio
try:
from twisted.internet.defer import Deferred
from twisted.python.failure import Failure
from twisted.web.iweb import IRequest
except ImportError:
pass
@@ -136,12 +136,11 @@ class HttpTransactionCache:
# if the request fails with an exception, remove it
# from the transaction map. This is done to ensure that we don't
# cache transient errors like rate-limiting errors, etc.
def remove_from_map(err: Failure) -> None:
self.transactions.pop(txn_key, None)
# we deliberately do not propagate the error any further, as we
# expect the observers to have reported it.
def remove_from_map(t: asyncio.Task) -> None:
if t.cancelled() or t.exception() is not None:
self.transactions.pop(txn_key, None)
deferred.addErrback(remove_from_map)
deferred.add_done_callback(remove_from_map)
return make_deferred_yieldable(observable.observe())
synapse/storage/controllers/persist_events.py
+3 -5
View File
@@ -40,10 +40,7 @@ from typing import (
import attr
from prometheus_client import Counter, Histogram
try:
from twisted.internet import defer
except ImportError:
pass
import asyncio
from synapse.api.constants import EventTypes, Membership
from synapse.events import EventBase
@@ -232,8 +229,9 @@ class _EventPeristenceQueue(Generic[_PersistResult]):
# the new task has been merged into the last task in the queue
end_item = queue[-1]
else:
loop = asyncio.get_running_loop()
deferred: ObservableDeferred[_PersistResult] = ObservableDeferred(
defer.Deferred(), consumeErrors=True
loop.create_future(), consumeErrors=True
)
end_item = _EventPersistQueueItem(
synapse/storage/databases/main/events_worker.py
+30 -27
View File
@@ -38,10 +38,7 @@ from typing import (
import attr
from prometheus_client import Gauge
try:
from twisted.internet import defer
except ImportError:
pass
import asyncio
from synapse.api.constants import Direction, EventTypes
from synapse.api.errors import NotFoundError, SynapseError
@@ -89,7 +86,6 @@ from synapse.storage.util.sequence import build_sequence_generator
from synapse.types import JsonDict, get_domain_from_id
from synapse.types.state import StateFilter
from synapse.types.storage import _BackgroundUpdates
from synapse.util import unwrapFirstError
from synapse.util.async_helpers import ObservableDeferred, delay_cancellation
from synapse.util.caches.descriptors import cached, cachedList
from synapse.util.caches.lrucache import AsyncLruCache
@@ -316,7 +312,7 @@ class EventsWorkerStore(SQLBaseStore):
self._event_fetch_lock = threading.Condition()
self._event_fetch_list: list[
tuple[Iterable[str], "defer.Deferred[dict[str, _EventRow]]"]
tuple[Iterable[str], "asyncio.Future[dict[str, _EventRow]]"]
] = []
self._event_fetch_ongoing = 0
event_fetch_ongoing_gauge.labels(**{SERVER_NAME_LABEL: self.server_name}).set(
@@ -873,8 +869,10 @@ class EventsWorkerStore(SQLBaseStore):
# to all the events we pulled from the DB (this will result in this
# function returning more events than requested, but that can happen
# already due to `_get_events_from_db`).
loop = asyncio.get_event_loop()
_fetching_future: asyncio.Future[dict[str, EventCacheEntry]] = loop.create_future()
fetching_deferred: ObservableDeferred[dict[str, EventCacheEntry]] = (
ObservableDeferred(defer.Deferred(), consumeErrors=True)
ObservableDeferred(_fetching_future, consumeErrors=True)
)
for event_id in missing_events_ids:
self._current_event_fetches[event_id] = fetching_deferred
@@ -897,7 +895,8 @@ class EventsWorkerStore(SQLBaseStore):
missing_events.update(db_missing_events)
except Exception as e:
with PreserveLoggingContext():
fetching_deferred.errback(e)
if not _fetching_future.done():
_fetching_future.set_exception(e)
raise e
finally:
# Ensure that we mark these events as no longer being fetched.
@@ -905,7 +904,8 @@ class EventsWorkerStore(SQLBaseStore):
self._current_event_fetches.pop(event_id, None)
with PreserveLoggingContext():
fetching_deferred.callback(missing_events)
if not _fetching_future.done():
_fetching_future.set_result(missing_events)
return missing_events
@@ -920,12 +920,9 @@ class EventsWorkerStore(SQLBaseStore):
if already_fetching_deferreds:
# Wait for the other event requests to finish and add their results
# to ours.
results = await make_deferred_yieldable(
defer.gatherResults(
(d.observe() for d in already_fetching_deferreds),
consumeErrors=True,
)
).addErrback(unwrapFirstError)
results = await asyncio.gather(
*(d.observe() for d in already_fetching_deferreds),
)
for result in results:
# We filter out events that we haven't asked for as we might get
@@ -1234,8 +1231,9 @@ class EventsWorkerStore(SQLBaseStore):
# Fail any outstanding fetches since no one else will handle them.
assert exc is not None
with PreserveLoggingContext():
for _, deferred in event_fetches_to_fail:
deferred.errback(exc)
for _, future in event_fetches_to_fail:
if not future.done():
future.set_exception(exc)
def _fetch_loop(self, conn: LoggingDatabaseConnection) -> None:
"""Takes a database connection and waits for requests for events from
@@ -1269,7 +1267,7 @@ class EventsWorkerStore(SQLBaseStore):
def _fetch_event_list(
self,
conn: LoggingDatabaseConnection,
event_list: list[tuple[Iterable[str], "defer.Deferred[dict[str, _EventRow]]"]],
event_list: list[tuple[Iterable[str], "asyncio.Future[dict[str, _EventRow]]"]],
) -> None:
"""Handle a load of requests from the _event_fetch_list queue
@@ -1303,23 +1301,27 @@ class EventsWorkerStore(SQLBaseStore):
events_to_fetch,
)
# We only want to resolve deferreds from the main thread
# We only want to resolve futures from the main thread
loop = asyncio.get_event_loop()
def fire() -> None:
for _, d in event_list:
d.callback(row_dict)
if not d.done():
d.set_result(row_dict)
with PreserveLoggingContext():
self.hs.get_reactor().callFromThread(fire)
loop.call_soon_threadsafe(fire)
except Exception as e:
logger.exception("do_fetch")
# We only want to resolve deferreds from the main thread
# We only want to resolve futures from the main thread
loop = asyncio.get_event_loop()
def fire_errback(exc: Exception) -> None:
for _, d in event_list:
d.errback(exc)
if not d.done():
d.set_exception(exc)
with PreserveLoggingContext():
self.hs.get_reactor().callFromThread(fire_errback, e)
loop.call_soon_threadsafe(fire_errback, e)
@trace
async def _get_events_from_db(
@@ -1544,7 +1546,8 @@ class EventsWorkerStore(SQLBaseStore):
that weren't requested.
"""
events_d: "defer.Deferred[dict[str, _EventRow]]" = defer.Deferred()
loop = asyncio.get_running_loop()
events_d: asyncio.Future[dict[str, _EventRow]] = loop.create_future()
with self._event_fetch_lock:
self._event_fetch_list.append((events, events_d))
self._event_fetch_lock.notify()
synapse/storage/databases/main/push_rule.py
+20 -32
View File
@@ -29,14 +29,9 @@ from typing import (
cast,
)
try:
from twisted.internet import defer
except ImportError:
pass
from synapse.api.errors import StoreError
from synapse.config.homeserver import ExperimentalConfig
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.context import run_in_background
from synapse.replication.tcp.streams import PushRulesStream
from synapse.storage._base import SQLBaseStore
from synapse.storage.database import (
@@ -54,7 +49,6 @@ from synapse.storage.push_rule import InconsistentRuleException, RuleNotFoundExc
from synapse.storage.util.id_generators import IdGenerator, MultiWriterIdGenerator
from synapse.synapse_rust.push import FilteredPushRules, PushRule, PushRules
from synapse.types import JsonDict
from synapse.util import unwrapFirstError
from synapse.util.async_helpers import gather_results
from synapse.util.caches.descriptors import cached, cachedList
from synapse.util.caches.stream_change_cache import StreamChangeCache
@@ -273,33 +267,27 @@ class PushRulesWorkerStore(
user_id: [] for user_id in user_ids
}
# gatherResults loses all type information.
rows, enabled_map_by_user = await make_deferred_yieldable(
gather_results(
(
cast(
Any,
run_in_background(
self.db_pool.simple_select_many_batch,
table="push_rules",
column="user_name",
iterable=user_ids,
retcols=(
"user_name",
"rule_id",
"priority_class",
"priority",
"conditions",
"actions",
),
desc="bulk_get_push_rules",
batch_size=1000,
),
# gather_results loses all type information.
rows, enabled_map_by_user = await gather_results(
(
run_in_background(
self.db_pool.simple_select_many_batch,
table="push_rules",
column="user_name",
iterable=user_ids,
retcols=(
"user_name",
"rule_id",
"priority_class",
"priority",
"conditions",
"actions",
),
run_in_background(self.bulk_get_push_rules_enabled, user_ids),
desc="bulk_get_push_rules",
batch_size=1000,
),
consumeErrors=True,
).addErrback(unwrapFirstError)
run_in_background(self.bulk_get_push_rules_enabled, user_ids),
),
)
# Sort by highest priority_class, then highest priority.
synapse/storage/databases/main/stream.py
+14 -20
View File
@@ -58,15 +58,12 @@ from typing import (
import attr
from typing_extensions import assert_never
try:
from twisted.internet import defer
except ImportError:
pass
import asyncio
from synapse.api.constants import Direction, EventTypes, Membership
from synapse.api.filtering import Filter
from synapse.events import EventBase
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.context import run_in_background
from synapse.logging.opentracing import tag_args, trace
from synapse.storage._base import SQLBaseStore
from synapse.storage.database import (
@@ -704,21 +701,18 @@ class StreamWorkerStore(EventsWorkerStore, SQLBaseStore):
results = {}
room_ids = list(room_ids)
for rm_ids in (room_ids[i : i + 20] for i in range(0, len(room_ids), 20)):
res = await make_deferred_yieldable(
defer.gatherResults(
[
run_in_background(
self.paginate_room_events_by_stream_ordering,
room_id=room_id,
from_key=from_key,
to_key=to_key,
direction=direction,
limit=limit,
)
for room_id in rm_ids
],
consumeErrors=True,
)
res = await asyncio.gather(
*(
run_in_background(
self.paginate_room_events_by_stream_ordering,
room_id=room_id,
from_key=from_key,
to_key=to_key,
direction=direction,
limit=limit,
)
for room_id in rm_ids
),
)
results.update(dict(zip(rm_ids, res)))
synapse/storage/util/partial_state_events_tracker.py
+20 -27
View File
@@ -19,21 +19,15 @@
#
#
import asyncio
import logging
from collections import defaultdict
from typing import Collection
try:
from twisted.internet import defer
from twisted.internet.defer import Deferred
except ImportError:
pass
from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable
from synapse.logging.context import PreserveLoggingContext
from synapse.logging.opentracing import trace_with_opname
from synapse.storage.databases.main.events_worker import EventsWorkerStore
from synapse.storage.databases.main.room import RoomWorkerStore
from synapse.util import unwrapFirstError
from synapse.util.cancellation import cancellable
logger = logging.getLogger(__name__)
@@ -44,9 +38,9 @@ class PartialStateEventsTracker:
def __init__(self, store: EventsWorkerStore):
self._store = store
# a map from event id to a set of Deferreds which are waiting for that event to be
# a map from event id to a set of Futures which are waiting for that event to be
# un-partial-stated.
self._observers: dict[str, set[Deferred[None]]] = defaultdict(set)
self._observers: dict[str, set[asyncio.Future[None]]] = defaultdict(set)
def notify_un_partial_stated(self, event_id: str) -> None:
"""Notify that we now have full state for a given event
@@ -68,7 +62,8 @@ class PartialStateEventsTracker:
)
with PreserveLoggingContext():
for o in observers:
o.callback(None)
if not o.done():
o.set_result(None)
@trace_with_opname("PartialStateEventsTracker.await_full_state")
@cancellable
@@ -96,8 +91,9 @@ class PartialStateEventsTracker:
)
# create an observer for each lazy-joined event
observers: dict[str, Deferred[None]] = {
event_id: Deferred() for event_id in partial_state_event_ids
loop = asyncio.get_event_loop()
observers: dict[str, asyncio.Future[None]] = {
event_id: loop.create_future() for event_id in partial_state_event_ids
}
for event_id, observer in observers.items():
self._observers[event_id].add(observer)
@@ -111,15 +107,10 @@ class PartialStateEventsTracker:
).items():
# there may have been a call to notify_un_partial_stated during the
# db query, so the observers may already have been called.
if not partial and not observers[event_id].called:
observers[event_id].callback(None)
if not partial and not observers[event_id].done():
observers[event_id].set_result(None)
await make_deferred_yieldable(
defer.gatherResults(
observers.values(),
consumeErrors=True,
)
).addErrback(unwrapFirstError)
await asyncio.gather(*observers.values())
logger.info("Events %s all un-partial-stated", observers.keys())
finally:
# remove any observers we created. This should happen when the notification
@@ -141,9 +132,9 @@ class PartialCurrentStateTracker:
def __init__(self, store: RoomWorkerStore):
self._store = store
# a map from room id to a set of Deferreds which are waiting for that room to be
# a map from room id to a set of Futures which are waiting for that room to be
# un-partial-stated.
self._observers: dict[str, set[Deferred[None]]] = defaultdict(set)
self._observers: dict[str, set[asyncio.Future[None]]] = defaultdict(set)
def notify_un_partial_stated(self, room_id: str) -> None:
"""Notify that we now have full current state for a given room
@@ -163,14 +154,16 @@ class PartialCurrentStateTracker:
)
with PreserveLoggingContext():
for o in observers:
o.callback(None)
if not o.done():
o.set_result(None)
@trace_with_opname("PartialCurrentStateTracker.await_full_state")
@cancellable
async def await_full_state(self, room_id: str) -> None:
# We add the deferred immediately so that the DB call to check for
# We add the future immediately so that the DB call to check for
# partial state doesn't race when we unpartial the room.
d: Deferred[None] = Deferred()
loop = asyncio.get_event_loop()
d: asyncio.Future[None] = loop.create_future()
self._observers.setdefault(room_id, set()).add(d)
try:
@@ -185,7 +178,7 @@ class PartialCurrentStateTracker:
stack_info=True,
)
await make_deferred_yieldable(d)
await d
logger.info("Room has un-partial-stated")
finally:
synapse/util/async_helpers.py
+18
View File
@@ -421,6 +421,24 @@ class ObservableFuture(Generic[_T]):
raise ValueError(f"{self!r} has no result yet")
return self._result[1]
def callback(self, result: _T) -> None:
"""Resolve the underlying future with a result (Deferred compat)."""
self._future.set_result(result)
def errback(self, exc: BaseException | None = None) -> None:
"""Reject the underlying future with an exception (Deferred compat).
If no exception is given, uses the current exception from sys.exc_info().
"""
if exc is None:
import sys
_, exc_val, _ = sys.exc_info()
if exc_val is None:
raise RuntimeError("errback() called with no active exception")
exc = exc_val
self._future.set_exception(exc)
class NativeLinearizer:
"""asyncio-native equivalent of Linearizer.
synapse/util/batching_queue.py
+11 -19
View File
@@ -31,12 +31,9 @@ from typing import (
from prometheus_client import Gauge
try:
from twisted.internet import defer
except ImportError:
pass
import asyncio
from synapse.logging.context import PreserveLoggingContext, make_deferred_yieldable
from synapse.logging.context import PreserveLoggingContext
from synapse.metrics import SERVER_NAME_LABEL
from synapse.util.clock import Clock
from synapse.util.duration import Duration
@@ -149,15 +146,10 @@ class BatchingQueue(Generic[V, R]):
key.
"""
# First we create a defer and add it and the value to the list of
# First we create a future and add it and the value to the list of
# pending items.
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()
loop = asyncio.get_running_loop()
d: asyncio.Future[V] = loop.create_future()
self._next_values.setdefault(key, []).append((value, d))
# If we're not currently processing the key fire off a background
@@ -166,7 +158,7 @@ class BatchingQueue(Generic[V, R]):
self.hs.run_as_background_process(self._name, self._process_queue, key)
with self._number_in_flight_metric.track_inprogress():
return await make_deferred_yieldable(d)
return await d
async def _process_queue(self, key: Hashable) -> None:
"""A background task to repeatedly pull things off the queue for the
@@ -197,16 +189,16 @@ class BatchingQueue(Generic[V, R]):
results = await self._process_batch_callback(values)
with PreserveLoggingContext():
for _, deferred in next_values:
deferred.callback(results)
for _, future in next_values:
future.set_result(results)
except Exception as e:
with PreserveLoggingContext():
for _, deferred in next_values:
if deferred.called:
for _, future in next_values:
if future.done():
continue
deferred.errback(e)
future.set_exception(e)
finally:
self._processing_keys.discard(key)
synapse/util/caches/cached_call.py
+24 -25
View File
@@ -18,15 +18,10 @@
# [This file includes modifications made by New Vector Limited]
#
#
import asyncio
import enum
from typing import Awaitable, Callable, Generic, TypeVar
try:
from twisted.internet.defer import Deferred
from twisted.python.failure import Failure
except ImportError:
pass
from synapse.logging.context import make_deferred_yieldable, run_in_background
TV = TypeVar("TV")
@@ -69,7 +64,7 @@ class CachedCall(Generic[TV]):
CachedCall object.
"""
__slots__ = ["_callable", "_deferred", "_result"]
__slots__ = ["_callable", "_task", "_result", "_exception"]
def __init__(self, f: Callable[[], Awaitable[TV]]):
"""
@@ -78,28 +73,33 @@ class CachedCall(Generic[TV]):
at once (per instance of CachedCall)
"""
self._callable: Callable[[], Awaitable[TV]] | None = f
self._deferred: Deferred | None = None
self._result: _Sentinel | TV | Failure = _Sentinel.sentinel
self._task: asyncio.Task | None = None
self._result: _Sentinel | TV = _Sentinel.sentinel
self._exception: BaseException | None = None
async def get(self) -> TV:
"""Kick off the call if necessary, and return the result"""
# Fire off the callable now if this is our first time
if not self._deferred:
if not self._task:
assert self._callable is not None
self._deferred = run_in_background(self._callable)
self._task = run_in_background(self._callable)
# we will never need the callable again, so make sure it can be GCed
self._callable = None
# once the deferred completes, store the result. We cannot simply leave the
# result in the deferred, since `awaiting` a deferred destroys its result.
# (Also, if it's a Failure, GCing the deferred would log a critical error
# about unhandled Failures)
def got_result(r: TV | Failure) -> None:
self._result = r
# once the task completes, store the result.
def got_result(t: asyncio.Task) -> None:
if t.cancelled():
self._exception = asyncio.CancelledError()
return
exc = t.exception()
if exc is not None:
self._exception = exc
else:
self._result = t.result()
self._deferred.addBoth(got_result)
self._task.add_done_callback(got_result)
# TODO: consider cancellation semantics. Currently, if the call to get()
# is cancelled, the underlying call will continue (and any future calls
@@ -107,15 +107,14 @@ class CachedCall(Generic[TV]):
# the fact the underlying call may be logged to a cancelled logcontext,
# and any eventual exception may not be reported.
# we can now await the deferred, and once it completes, return the result.
if isinstance(self._result, _Sentinel):
await make_deferred_yieldable(self._deferred)
assert not isinstance(self._result, _Sentinel)
# we can now await the task, and once it completes, return the result.
if isinstance(self._result, _Sentinel) and self._exception is None:
await make_deferred_yieldable(self._task)
if isinstance(self._result, Failure):
self._result.raiseException()
raise AssertionError("unexpected return from Failure.raiseException")
if self._exception is not None:
raise self._exception
assert not isinstance(self._result, _Sentinel)
return self._result
synapse/util/caches/response_cache.py
+3 -9
View File
@@ -31,11 +31,6 @@ from typing import (
import attr
try:
from twisted.internet import defer
except ImportError:
pass
from synapse.logging.context import make_deferred_yieldable, run_in_background
from synapse.logging.opentracing import (
active_span,
@@ -237,7 +232,7 @@ class ResponseCache(Generic[KV]):
)
self._result_cache[key] = entry
def on_complete(r: RV) -> RV:
def on_complete(f: Any) -> None:
# if this cache has a non-zero timeout, and the callback has not cleared
# the should_cache bit, we leave it in the cache for now and schedule
# its removal later.
@@ -256,12 +251,11 @@ class ResponseCache(Generic[KV]):
else:
# otherwise, remove the result immediately.
self.unset(key)
return r
# make sure we do this *after* adding the entry to result_cache,
# in case the result is already complete (in which case flipping the order would
# leave us with a stuck entry in the cache).
result.addBoth(on_complete)
deferred.add_done_callback(on_complete)
return entry
def unset(self, key: KV) -> None:
@@ -388,7 +382,7 @@ class ResponseCache(Generic[KV]):
result = entry.result.observe()
if self._enable_logging:
if result.called:
if entry.result.has_called():
logger.info(
"[%s]: using completed cached result for [%s]", self._name, key
)
synapse/util/clock.py
+10 -3
View File
@@ -629,6 +629,11 @@ class NativeLoopingCall:
def __init__(self, task: "asyncio.Task[None]") -> None:
self._task = task
@property
def running(self) -> bool:
"""Whether the looping call is still running."""
return not self._task.done()
def stop(self) -> None:
"""Stop the looping call."""
self._task.cancel()
@@ -725,7 +730,8 @@ class NativeClock:
# 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._pending_sleeps: list[tuple[float, int, asyncio.Future]] = []
self._sleep_seq = 0 # tiebreaker for heapq when wake_times are equal
self._use_fake_time = False # Set to True by tests
def _get_loop(self) -> asyncio.AbstractEventLoop:
@@ -758,7 +764,8 @@ class NativeClock:
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))
self._sleep_seq += 1
heapq.heappush(self._pending_sleeps, (wake_time, self._sleep_seq, future))
await future
else:
await asyncio.sleep(duration.as_secs())
@@ -774,7 +781,7 @@ class NativeClock:
# 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)
_, _, future = heapq.heappop(self._pending_sleeps)
if not future.done():
future.set_result(None)
tests/server.py
+16 -3
View File
@@ -156,6 +156,7 @@ class FakeChannel:
_producer: Optional[Union[IPullProducer, IPushProducer]] = None
resource_usage: ContextResourceUsage | None = None
_request: Request | None = None
_clock: Any = None # NativeClock, for advancing fake time
@property
def request(self) -> Request:
@@ -319,10 +320,21 @@ class FakeChannel:
if _time.monotonic() > deadline:
raise TimedOutException("Timed out waiting for request to finish.")
# Advance NativeClock fake time (fires pending sleeps)
if self._clock is not None:
self._clock.advance(0.0)
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))
# Use a small real sleep to allow thread pool callbacks to be delivered.
if not loop.is_closed():
async def _drain() -> None:
"""Run multiple event loop ticks to drain pending work."""
for _ in range(20):
await asyncio.sleep(0.001)
if self._clock is not None:
self._clock.advance(0.0)
loop.run_until_complete(_drain())
def extract_cookies(self, cookies: MutableMapping[str, str]) -> None:
"""Process the contents of any Set-Cookie headers in the response
@@ -384,6 +396,7 @@ def make_request(
await_result: bool = True,
custom_headers: Iterable[CustomHeaderType] | None = None,
client_ip: str = "127.0.0.1",
clock: Any = None,
) -> FakeChannel:
"""
Make a web request using the given method, path and content, and render it
@@ -440,7 +453,7 @@ def make_request(
if isinstance(content, str):
content = content.encode("utf8")
channel = FakeChannel(site, reactor, ip=client_ip)
channel = FakeChannel(site, reactor, ip=client_ip, clock=clock)
req = request(
channel,
tests/unittest.py
+1
View File
@@ -803,6 +803,7 @@ class HomeserverTestCase(TestCase):
await_result,
custom_headers,
client_ip,
clock=self.clock,
)
def setup_test_homeserver(