Experimental Federation Speedup

changelog.d/9702.misc

@@ -0,0 +1 @@
+Speed up Federation by using less database calls.

synapse/federation/sender/__init__.py

@@ -40,7 +40,7 @@
     events_processed_counter,
 )
 from synapse.metrics.background_process_metrics import run_as_background_process
-from synapse.types import JsonDict, ReadReceipt, RoomStreamToken
+from synapse.types import Collection, JsonDict, ReadReceipt, RoomStreamToken
 from synapse.util.metrics import Measure, measure_func
 
 if TYPE_CHECKING:
@@ -275,15 +275,17 @@ async def _process_event_queue_loop(self) -> None:
                 if not events and next_token >= self._last_poked_id:
                     break
 
-                async def handle_event(event: EventBase) -> None:
+                async def handle_event(
+                    event: EventBase,
+                ) -> Optional[Tuple[EventBase, Collection[str]]]:
                     # Only send events for this server.
                     send_on_behalf_of = event.internal_metadata.get_send_on_behalf_of()
                     is_mine = self.is_mine_id(event.sender)
                     if not is_mine and send_on_behalf_of is None:
-                        return
+                        return None
 
                     if not event.internal_metadata.should_proactively_send():
-                        return
+                        return None
 
                     destinations = None  # type: Optional[Set[str]]
                     if not event.prev_event_ids():
@@ -318,7 +320,7 @@ async def handle_event(event: EventBase) -> None:
                                 "Failed to calculate hosts in room for event: %s",
                                 event.event_id,
                             )
-                            return
+                            return None
 
                     destinations = {
                         d
@@ -328,6 +330,8 @@ async def handle_event(event: EventBase) -> None:
                         )
                     }
 
+                    destinations.discard(self.server_name)
+
                     if send_on_behalf_of is not None:
                         # If we are sending the event on behalf of another server
                         # then it already has the event and there is no reason to
@@ -337,34 +341,84 @@ async def handle_event(event: EventBase) -> None:
                     logger.debug("Sending %s to %r", event, destinations)
 
                     if destinations:
-                        await self._send_pdu(event, destinations)
-
                         now = self.clock.time_msec()
                         ts = await self.store.get_received_ts(event.event_id)
 
                         synapse.metrics.event_processing_lag_by_event.labels(
                             "federation_sender"
                         ).observe((now - ts) / 1000)
 
-                async def handle_room_events(events: Iterable[EventBase]) -> None:
+                        return event, destinations
+                    return None
+
+                async def handle_room_events(
+                    room_id: str, events: Iterable[EventBase]
+                ) -> Tuple[str, Dict[EventBase, Collection[str]]]:
                     with Measure(self.clock, "handle_room_events"):
-                        for event in events:
-                            await handle_event(event)
+                        # Handle all events, skip if handle_event returns None
+                        handled_events = (
+                            ret
+                            for ret in [await handle_event(event) for event in events]
+                            if ret is not None
+                        )
+
+                        # Generate event -> destination pairs for every handled event
+                        return room_id, {
+                            event: dests for event, dests in handled_events
+                        }
 
                 events_by_room = {}  # type: Dict[str, List[EventBase]]
                 for event in events:
                     events_by_room.setdefault(event.room_id, []).append(event)
 
-                await make_deferred_yieldable(
+                per_room_events_and_dests = await make_deferred_yieldable(
                     defer.gatherResults(
                         [
-                            run_in_background(handle_room_events, evs)
-                            for evs in events_by_room.values()
+                            run_in_background(handle_room_events, rid, evs)
+                            for rid, evs in events_by_room.items()
                         ],
                         consumeErrors=True,
                     )
+                )  # type: List[Tuple[str, Dict[EventBase, Collection[str]]]]
+
+                # Tuples of room_id + destination to their max-seen stream_ordering
+                room_with_dest_stream_ordering = {}  # type: Dict[Tuple[str, str], int]
+
+                # List of events to send to each destination
+                event_to_dests = {}  # type: Dict[str, List[EventBase]]
+
+                # For each roomID + events with destinations pair...
+                for room_id, event_and_dests in per_room_events_and_dests:
+                    # ...get every event with their destinations...
+                    for event, destinations in event_and_dests.items():
+
+                        # (we got this from the database, it's filled)
+                        assert event.internal_metadata.stream_ordering
+
+                        # ...iterate over those destinations..
+                        for destination in destinations:
+                            if (
+                                room_with_dest_stream_ordering.setdefault(
+                                    (room_id, destination), 0
+                                )
+                                < event.internal_metadata.stream_ordering
+                            ):
+                                # ...update their stream-ordering...
+                                room_with_dest_stream_ordering[
+                                    (room_id, destination)
+                                ] = event.internal_metadata.stream_ordering
+
+                            # ...and add the event to each destination queue.
+                            event_to_dests.setdefault(destination, []).append(event)
+
+                # Bulk-store destination_rooms stream_ids
+                await self.store.bulk_store_destination_rooms_entries(
+                    room_with_dest_stream_ordering
                 )
 
+                # Send corresponding events to each destination queue
+                self._distribute_pdus(event_to_dests)
+
                 await self.store.update_federation_out_pos("events", next_token)
 
                 if events:
@@ -393,34 +447,13 @@ async def handle_room_events(events: Iterable[EventBase]) -> None:
         finally:
             self._is_processing = False
 
-    async def _send_pdu(self, pdu: EventBase, destinations: Iterable[str]) -> None:
-        # We loop through all destinations to see whether we already have
-        # a transaction in progress. If we do, stick it in the pending_pdus
-        # table and we'll get back to it later.
+    def _distribute_pdus(self, destination_to_pdus: Dict[str, List[EventBase]]) -> None:
+        for destination, pdus in destination_to_pdus.items():
+            logger.debug("Sending %d to: %s", len(pdus), destination)
 
-        destinations = set(destinations)
-        destinations.discard(self.server_name)
-        logger.debug("Sending to: %s", str(destinations))
+            sent_pdus_destination_dist_total.inc(len(pdus))
 
-        if not destinations:
-            return
-
-        sent_pdus_destination_dist_total.inc(len(destinations))
-        sent_pdus_destination_dist_count.inc()
-
-        assert pdu.internal_metadata.stream_ordering
-
-        # track the fact that we have a PDU for these destinations,
-        # to allow us to perform catch-up later on if the remote is unreachable
-        # for a while.
-        await self.store.store_destination_rooms_entries(
-            destinations,
-            pdu.room_id,
-            pdu.internal_metadata.stream_ordering,
-        )
-
-        for destination in destinations:
-            self._get_per_destination_queue(destination).send_pdu(pdu)
+            self._get_per_destination_queue(destination).send_pdu(*pdus)
 
     async def send_read_receipt(self, receipt: ReadReceipt) -> None:
         """Send a RR to any other servers in the room

synapse/federation/sender/per_destination_queue.py

@@ -154,7 +154,7 @@ def pending_edu_count(self) -> int:
             + len(self._pending_edus_keyed)
         )
 
-    def send_pdu(self, pdu: EventBase) -> None:
+    def send_pdu(self, *pdus: EventBase) -> None:
         """Add a PDU to the queue, and start the transmission loop if necessary
 
         Args:
@@ -163,10 +163,13 @@ def send_pdu(self, pdu: EventBase) -> None:
         if not self._catching_up or self._last_successful_stream_ordering is None:
             # only enqueue the PDU if we are not catching up (False) or do not
             # yet know if we have anything to catch up (None)
-            self._pending_pdus.append(pdu)
+            self._pending_pdus.extend(pdus)
         else:
-            assert pdu.internal_metadata.stream_ordering
-            self._catchup_last_skipped = pdu.internal_metadata.stream_ordering
+            self._catchup_last_skipped = max(
+                pdu.internal_metadata.stream_ordering
+                for pdu in pdus
+                if pdu.internal_metadata.stream_ordering is not None
+            )
 
         self.attempt_new_transaction()
 

synapse/storage/databases/main/transactions.py

@@ -15,7 +15,7 @@
 
 import logging
 from collections import namedtuple
-from typing import Iterable, List, Optional, Tuple
+from typing import Dict, List, Optional, Tuple
 
 from canonicaljson import encode_canonical_json
 
@@ -296,37 +296,33 @@ def _set_destination_retry_timings_emulated(
                 },
             )
 
-    async def store_destination_rooms_entries(
-        self,
-        destinations: Iterable[str],
-        room_id: str,
-        stream_ordering: int,
-    ) -> None:
+    async def bulk_store_destination_rooms_entries(
+        self, room_and_destination_to_ordering: Dict[Tuple[str, str], int]
+    ):
         """
-        Updates or creates `destination_rooms` entries in batch for a single event.
+        Updates or creates `destination_rooms` entries in batch for many events.
 
         Args:
-            destinations: list of destinations
-            room_id: the room_id of the event
-            stream_ordering: the stream_ordering of the event
+            room_and_destination_to_ordering: A mapping of (room, destination) -> stream_id
         """
 
         await self.db_pool.simple_upsert_many(
             table="destinations",
             key_names=("destination",),
-            key_values=[(d,) for d in destinations],
+            key_values=[(key[1],) for key in room_and_destination_to_ordering.keys()],
             value_names=[],
             value_values=[],
             desc="store_destination_rooms_entries_dests",
         )
 
-        rows = [(destination, room_id) for destination in destinations]
         await self.db_pool.simple_upsert_many(
             table="destination_rooms",
-            key_names=("destination", "room_id"),
-            key_values=rows,
+            key_names=("room_id", "destination"),
+            key_values=list(room_and_destination_to_ordering.keys()),
             value_names=["stream_ordering"],
-            value_values=[(stream_ordering,)] * len(rows),
+            value_values=[
+                (stream_id,) for stream_id in room_and_destination_to_ordering.values()
+            ],
             desc="store_destination_rooms_entries_rooms",
         )