Merge commit 'de119063f' into anoa/dinsic_release_1_18_x

* commit 'de119063f': (31 commits)
  Convert room list handler to async/await. (#7912)
  Element CSS and logo in email templates (#7919)
  Lint the contrib/ directory in CI and linting scripts, add synctl to linting script (#7914)
  Remove unused code from synapse.logging.utils. (#7897)
  Fix a typo in the sample config. (#7890)
  Fix deprecation warning: import ABC from collections.abc (#7892)
  Change sample config's postgres user to synapse_user (#7889)
  Fix deprecation warning due to invalid escape sequences (#7895)
  Remove Ubuntu Eoan that is now EOL (#7888)
  Fix the trace function for async functions. (#7872)
  Add help for creating a user via docker (#7885)
  Switch to Debian:Slim from Alpine for the docker image (#7839)
  Stop using 'device_max_stream_id' (#7882)
  Fix TypeError in synapse.notifier (#7880)
  Add a default limit (of 100) to get/sync operations. (#7858)
  Change "unknown room ver" logging to warning. (#7881)
  Convert device handler to async/await (#7871)
  Convert synapse.app to async/await. (#7868)
  Convert _base, profile, and _receipts handlers to async/await (#7860)
  Add admin endpoint to get members in a room. (#7842)
  ...

1 files changed, 36 insertions, 2 deletions

diff --git a/synapse/config/_base.py b/synapse/config/_base.py

index f2830c609d..34a2370e67 100644
--- a/synapse/config/_base.py
+++ b/synapse/config/_base.py
@@ -19,10 +19,12 @@ import argparse
 import errno
 import os
 from collections import OrderedDict
+from hashlib import sha256
 from io import open as io_open
 from textwrap import dedent
-from typing import Any, MutableMapping, Optional
+from typing import Any, List, MutableMapping, Optional
 
+import attr
 import yaml
 
 
@@ -718,4 +720,36 @@ def find_config_files(search_paths):
     return config_files
 
 
-__all__ = ["Config", "RootConfig"]
+@attr.s
+class ShardedWorkerHandlingConfig:
+    """Algorithm for choosing which instance is responsible for handling some
+    sharded work.
+
+    For example, the federation senders use this to determine which instances
+    handles sending stuff to a given destination (which is used as the `key`
+    below).
+    """
+
+    instances = attr.ib(type=List[str])
+
+    def should_handle(self, instance_name: str, key: str) -> bool:
+        """Whether this instance is responsible for handling the given key.
+        """
+
+        # If multiple instances are not defined we always return true.
+        if not self.instances or len(self.instances) == 1:
+            return True
+
+        # We shard by taking the hash, modulo it by the number of instances and
+        # then checking whether this instance matches the instance at that
+        # index.
+        #
+        # (Technically this introduces some bias and is not entirely uniform,
+        # but since the hash is so large the bias is ridiculously small).
+        dest_hash = sha256(key.encode("utf8")).digest()
+        dest_int = int.from_bytes(dest_hash, byteorder="little")
+        remainder = dest_int % (len(self.instances))
+        return self.instances[remainder] == instance_name
+
+
+__all__ = ["Config", "RootConfig", "ShardedWorkerHandlingConfig"]