Parallelise key storage

synapse/crypto/keyring.py

@@ -495,15 +495,20 @@ class Keyring(object):
         response_keys.update(verify_keys)
         response_keys.update(old_verify_keys)
 
-        for key_id in updated_key_ids:
+        yield defer.gatherResults(
-            yield self.store.store_server_keys_json(
+            [
-                server_name=server_name,
+                self.store.store_server_keys_json(
-                key_id=key_id,
+                    server_name=server_name,
-                from_server=server_name,
+                    key_id=key_id,
-                ts_now_ms=time_now_ms,
+                    from_server=server_name,
-                ts_expires_ms=ts_valid_until_ms,
+                    ts_now_ms=time_now_ms,
-                key_json_bytes=signed_key_json_bytes,
+                    ts_expires_ms=ts_valid_until_ms,
-            )
+                    key_json_bytes=signed_key_json_bytes,
+                )
+                for key_id in updated_key_ids
+            ],
+            consumeErrors=True,
+        ).addErrback(unwrapFirstError)
 
         results[server_name] = response_keys
 
@@ -591,8 +596,13 @@ class Keyring(object):
         Returns:
             A deferred that completes when the keys are stored.
         """
-        for key_id, key in verify_keys.items():
+        # TODO(markjh): Store whether the keys have expired.
-            # TODO(markjh): Store whether the keys have expired.
+        yield defer.gatherResults(
-            yield self.store.store_server_verify_key(
+            [
-                server_name, server_name, key.time_added, key
+                self.store.store_server_verify_key(
-            )
+                    server_name, server_name, key.time_added, key
+                )
+                for key_id, key in verify_keys.items()
+            ],
+            consumeErrors=True,
+        ).addErrback(unwrapFirstError)

synapse/storage/events.py

@@ -78,7 +78,7 @@ class EventsStore(SQLBaseStore):
 
             for chunk in chunks:
                 # We can't easily parallelize these since different chunks
-                # might contain the same event. :()
+                # might contain the same event. :(
                 yield self.runInteraction(
                     "persist_events",
                     self._persist_events_txn,