Merge branch 'develop' into matrix-org-hotfixes

* develop:
  Don't unnecessarily start bg process in replication sending loop. (#8670)
  Don't unnecessarily start bg process while handling typing. (#8668)

changelog.d/8668.misc

@@ -0,0 +1 @@
+Reduce number of OpenTracing spans started.

changelog.d/8670.misc

@@ -0,0 +1 @@
+Reduce number of OpenTracing spans started.

synapse/handlers/typing.py

@@ -167,20 +167,25 @@ def process_replication_rows(
             now_typing = set(row.user_ids)
             self._room_typing[row.room_id] = row.user_ids
 
-            run_as_background_process(
-                "_handle_change_in_typing",
-                self._handle_change_in_typing,
-                row.room_id,
-                prev_typing,
-                now_typing,
-            )
+            if self.federation:
+                run_as_background_process(
+                    "_send_changes_in_typing_to_remotes",
+                    self._send_changes_in_typing_to_remotes,
+                    row.room_id,
+                    prev_typing,
+                    now_typing,
+                )
 
-    async def _handle_change_in_typing(
+    async def _send_changes_in_typing_to_remotes(
         self, room_id: str, prev_typing: Set[str], now_typing: Set[str]
     ):
         """Process a change in typing of a room from replication, sending EDUs
         for any local users.
         """
+
+        if not self.federation:
+            return
+
         for user_id in now_typing - prev_typing:
             if self.is_mine_id(user_id):
                 await self._push_remote(RoomMember(room_id, user_id), True)

synapse/replication/tcp/resource.py

@@ -117,6 +117,16 @@ def on_notifier_poke(self):
                 stream.discard_updates_and_advance()
             return
 
+        # We check up front to see if anything has actually changed, as we get
+        # poked because of changes that happened on other instances.
+        if all(
+            stream.last_token == stream.current_token(self._instance_name)
+            for stream in self.streams
+        ):
+            return
+
+        # If there are updates then we need to set this even if we're already
+        # looping, as the loop needs to know that he might need to loop again.
         self.pending_updates = True
 
         if self.is_looping: