From 80467bbac3be6e008b807793dfd27c733936c15c Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 22 Feb 2019 14:38:14 +0000 Subject: Fix state cache invalidation on workers --- synapse/storage/_base.py | 40 ++++++++++++++++++++++++++++++++++------ 1 file changed, 34 insertions(+), 6 deletions(-) (limited to 'synapse/storage/_base.py') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 3d895da43c..5a80eef211 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -1342,15 +1342,43 @@ class SQLBaseStore(object): changed """ for member in members_changed: - self.get_rooms_for_user_with_stream_ordering.invalidate((member,)) + self._attempt_to_invalidate_cache( + "get_rooms_for_user_with_stream_ordering", (member,), + ) for host in set(get_domain_from_id(u) for u in members_changed): - self.is_host_joined.invalidate((room_id, host)) - self.was_host_joined.invalidate((room_id, host)) + self._attempt_to_invalidate_cache( + "is_host_joined", (room_id, host,), + ) + self._attempt_to_invalidate_cache( + "was_host_joined", (room_id, host,), + ) + + self._attempt_to_invalidate_cache( + "get_users_in_room", (room_id,), + ) + self._attempt_to_invalidate_cache( + "get_room_summary", (room_id,), + ) + self._attempt_to_invalidate_cache( + "get_current_state_ids", (room_id,), + ) + + def _attempt_to_invalidate_cache(self, cache_name, key): + """Attempts to invalidate the cache of the given name, ignoring if the + cache doesn't exist. Mainly used for invalidating caches on workers, + where they may not have the cache. - self.get_users_in_room.invalidate((room_id,)) - self.get_room_summary.invalidate((room_id,)) - self.get_current_state_ids.invalidate((room_id,)) + Args: + cache_name (str) + key (tuple) + """ + try: + getattr(self, cache_name).invalidate(key) + except AttributeError: + # We probably haven't pulled in the cache in this worker, + # which is fine. + pass def _send_invalidation_to_replication(self, txn, cache_name, keys): """Notifies replication that given cache has been invalidated. -- cgit 1.4.1 From f191be822b265b50339e17b2d95125f43dc700b3 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff <1389908+richvdh@users.noreply.github.com> Date: Wed, 27 Feb 2019 10:21:49 +0000 Subject: Add database version to phonehome stats. (#4753) --- changelog.d/4753.misc | 1 + synapse/app/homeserver.py | 3 +++ synapse/storage/_base.py | 8 ++++++++ synapse/storage/engines/postgres.py | 25 +++++++++++++++++++++++++ synapse/storage/engines/sqlite.py | 9 +++++++++ 5 files changed, 46 insertions(+) create mode 100644 changelog.d/4753.misc (limited to 'synapse/storage/_base.py') diff --git a/changelog.d/4753.misc b/changelog.d/4753.misc new file mode 100644 index 0000000000..98532cc971 --- /dev/null +++ b/changelog.d/4753.misc @@ -0,0 +1 @@ +Add database version to phonehome stats. diff --git a/synapse/app/homeserver.py b/synapse/app/homeserver.py index 05a97979ec..e8b6cc3114 100755 --- a/synapse/app/homeserver.py +++ b/synapse/app/homeserver.py @@ -555,6 +555,9 @@ def run(hs): stats["memory_rss"] += process.memory_info().rss stats["cpu_average"] += int(process.cpu_percent(interval=None)) + stats["database_engine"] = hs.get_datastore().database_engine_name + stats["database_server_version"] = hs.get_datastore().get_server_version() + logger.info("Reporting stats to matrix.org: %s" % (stats,)) try: yield hs.get_simple_http_client().put_json( diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 5a80eef211..190be34fb1 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -1596,6 +1596,14 @@ class SQLBaseStore(object): return cls.cursor_to_dict(txn) + @property + def database_engine_name(self): + return self.database_engine.module.__name__ + + def get_server_version(self): + """Returns a string describing the server version number""" + return self.database_engine.server_version + class _RollbackButIsFineException(Exception): """ This exception is used to rollback a transaction without implying diff --git a/synapse/storage/engines/postgres.py b/synapse/storage/engines/postgres.py index 4004427c7b..dc3238501c 100644 --- a/synapse/storage/engines/postgres.py +++ b/synapse/storage/engines/postgres.py @@ -23,6 +23,7 @@ class PostgresEngine(object): self.module = database_module self.module.extensions.register_type(self.module.extensions.UNICODE) self.synchronous_commit = database_config.get("synchronous_commit", True) + self._version = None # unknown as yet def check_database(self, txn): txn.execute("SHOW SERVER_ENCODING") @@ -87,3 +88,27 @@ class PostgresEngine(object): """ txn.execute("SELECT nextval('state_group_id_seq')") return txn.fetchone()[0] + + @property + def server_version(self): + """Returns a string giving the server version. For example: '8.1.5' + + Returns: + string + """ + # note that this is a bit of a hack because it relies on on_new_connection + # having been called at least once. Still, that should be a safe bet here. + numver = self._version + assert numver is not None + + # https://www.postgresql.org/docs/current/libpq-status.html#LIBPQ-PQSERVERVERSION + if numver >= 100000: + return "%i.%i" % ( + numver / 10000, numver % 10000, + ) + else: + return "%i.%i.%i" % ( + numver / 10000, + (numver % 10000) / 100, + numver % 100, + ) diff --git a/synapse/storage/engines/sqlite.py b/synapse/storage/engines/sqlite.py index 059ab81055..1bcd5b99a4 100644 --- a/synapse/storage/engines/sqlite.py +++ b/synapse/storage/engines/sqlite.py @@ -70,6 +70,15 @@ class Sqlite3Engine(object): self._current_state_group_id += 1 return self._current_state_group_id + @property + def server_version(self): + """Gets a string giving the server version. For example: '3.22.0' + + Returns: + string + """ + return "%i.%i.%i" % self.module.sqlite_version_info + # Following functions taken from: https://github.com/coleifer/peewee -- cgit 1.4.1 From 6bb1c028f190d8ba561a6deaa474c060efb7c502 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Wed, 27 Feb 2019 10:28:37 +0000 Subject: Limit cache invalidation replication line length (#4748) --- changelog.d/4748.misc | 1 + synapse/replication/tcp/protocol.py | 17 ++++++++++++++++- synapse/storage/_base.py | 15 +++++++++++---- 3 files changed, 28 insertions(+), 5 deletions(-) create mode 100644 changelog.d/4748.misc (limited to 'synapse/storage/_base.py') diff --git a/changelog.d/4748.misc b/changelog.d/4748.misc new file mode 100644 index 0000000000..4dc18378e7 --- /dev/null +++ b/changelog.d/4748.misc @@ -0,0 +1 @@ +Improve replication performance by reducing cache invalidation traffic. diff --git a/synapse/replication/tcp/protocol.py b/synapse/replication/tcp/protocol.py index 0b3fe6cbf5..429471c345 100644 --- a/synapse/replication/tcp/protocol.py +++ b/synapse/replication/tcp/protocol.py @@ -268,7 +268,17 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver): if "\n" in string: raise Exception("Unexpected newline in command: %r", string) - self.sendLine(string.encode("utf-8")) + encoded_string = string.encode("utf-8") + + if len(encoded_string) > self.MAX_LENGTH: + raise Exception( + "Failed to send command %s as too long (%d > %d)" % ( + cmd.NAME, + len(encoded_string), self.MAX_LENGTH, + ) + ) + + self.sendLine(encoded_string) self.last_sent_command = self.clock.time_msec() @@ -361,6 +371,11 @@ class BaseReplicationStreamProtocol(LineOnlyReceiver): def id(self): return "%s-%s" % (self.name, self.conn_id) + def lineLengthExceeded(self, line): + """Called when we receive a line that is above the maximum line length + """ + self.send_error("Line length exceeded") + class ServerReplicationStreamProtocol(BaseReplicationStreamProtocol): VALID_INBOUND_COMMANDS = VALID_CLIENT_COMMANDS diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 190be34fb1..a0333d5309 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -30,6 +30,7 @@ from synapse.api.errors import StoreError from synapse.metrics.background_process_metrics import run_as_background_process from synapse.storage.engines import PostgresEngine, Sqlite3Engine from synapse.types import get_domain_from_id +from synapse.util import batch_iter from synapse.util.caches.descriptors import Cache from synapse.util.logcontext import LoggingContext, PreserveLoggingContext from synapse.util.stringutils import exception_to_unicode @@ -1327,10 +1328,16 @@ class SQLBaseStore(object): """ txn.call_after(self._invalidate_state_caches, room_id, members_changed) - keys = itertools.chain([room_id], members_changed) - self._send_invalidation_to_replication( - txn, _CURRENT_STATE_CACHE_NAME, keys, - ) + # We need to be careful that the size of the `members_changed` list + # isn't so large that it causes problems sending over replication, so we + # send them in chunks. + # Max line length is 16K, and max user ID length is 255, so 50 should + # be safe. + for chunk in batch_iter(members_changed, 50): + keys = itertools.chain([room_id], chunk) + self._send_invalidation_to_replication( + txn, _CURRENT_STATE_CACHE_NAME, keys, + ) def _invalidate_state_caches(self, room_id, members_changed): """Invalidates caches that are based on the current state, but does -- cgit 1.4.1