From f5a70e0d2e890adea53b3f6565a3bbe92512a506 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 11 Feb 2015 15:01:15 +0000 Subject: Add a cache for get_event --- synapse/storage/__init__.py | 3 +++ synapse/storage/_base.py | 24 +++++++++++++++++++++--- 2 files changed, 24 insertions(+), 3 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index a63c59a8a2..1170d8b6ec 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -164,6 +164,9 @@ class DataStore(RoomMemberStore, RoomStore, stream_ordering=None, is_new_state=True, current_state=None): + # Remove the any existing cache entries for the event_id + self._get_event_cache.pop(event.event_id) + # We purposefully do this first since if we include a `current_state` # key, we *want* to update the `current_state_events` table if current_state: diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index 3e1ab0a159..f13b8f4fad 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -19,6 +19,7 @@ from synapse.events import FrozenEvent from synapse.events.utils import prune_event from synapse.util.logutils import log_function from synapse.util.logcontext import PreserveLoggingContext, LoggingContext +from synapse.util.lrucache import LruCache from twisted.internet import defer @@ -128,6 +129,8 @@ class SQLBaseStore(object): self._txn_perf_counters = PerformanceCounters() self._get_event_counters = PerformanceCounters() + self._get_event_cache = LruCache(hs.config.event_cache_size) + def start_profiling(self): self._previous_loop_ts = self._clock.time_msec() @@ -579,6 +582,20 @@ class SQLBaseStore(object): def _get_event_txn(self, txn, event_id, check_redacted=True, get_prev_content=False, allow_rejected=False): + + start_time = time.time() * 1000 + update_counter = self._get_event_counters.update + + try: + cache = self._get_event_cache.setdefault(event_id, {}) + # Separate cache entries for each way to invoke _get_event_txn + return cache[(check_redacted, get_prev_content, allow_rejected)] + except KeyError: + pass + finally: + start_time = update_counter("event_cache", start_time) + + sql = ( "SELECT e.internal_metadata, e.json, r.event_id, rej.reason " "FROM event_json as e " @@ -588,7 +605,6 @@ class SQLBaseStore(object): "LIMIT 1 " ) - start_time = time.time() * 1000 txn.execute(sql, (event_id,)) @@ -599,14 +615,16 @@ class SQLBaseStore(object): internal_metadata, js, redacted, rejected_reason = res - self._get_event_counters.update("select_event", start_time) + start_time = update_counter("select_event", start_time) if allow_rejected or not rejected_reason: - return self._get_event_from_row_txn( + result = self._get_event_from_row_txn( txn, internal_metadata, js, redacted, check_redacted=check_redacted, get_prev_content=get_prev_content, ) + cache[(check_redacted, get_prev_content, allow_rejected)] = result + return result else: return None -- cgit 1.4.1 From aff892ce79b9cd5060d268eefb5876fc7caf30f0 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 11 Feb 2015 15:02:35 +0000 Subject: Fix formatting --- synapse/storage/_base.py | 2 -- 1 file changed, 2 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py index f13b8f4fad..29fc334f45 100644 --- a/synapse/storage/_base.py +++ b/synapse/storage/_base.py @@ -595,7 +595,6 @@ class SQLBaseStore(object): finally: start_time = update_counter("event_cache", start_time) - sql = ( "SELECT e.internal_metadata, e.json, r.event_id, rej.reason " "FROM event_json as e " @@ -605,7 +604,6 @@ class SQLBaseStore(object): "LIMIT 1 " ) - txn.execute(sql, (event_id,)) res = txn.fetchone() -- cgit 1.4.1 From ddc25cf4e2df10e2b175e70a2371a7e24081572f Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 11 Feb 2015 15:23:28 +0000 Subject: Invalidate the cache for an event if it is redacted --- synapse/storage/__init__.py | 2 ++ 1 file changed, 2 insertions(+) (limited to 'synapse/storage') diff --git a/synapse/storage/__init__.py b/synapse/storage/__init__.py index 1170d8b6ec..a33e2298f6 100644 --- a/synapse/storage/__init__.py +++ b/synapse/storage/__init__.py @@ -423,6 +423,8 @@ class DataStore(RoomMemberStore, RoomStore, ) def _store_redaction(self, txn, event): + # invalidate the cache for the redacted event + self._get_event_cache.pop(event.redacts) txn.execute( "INSERT OR IGNORE INTO redactions " "(event_id, redacts) VALUES (?,?)", -- cgit 1.4.1 From 7be0f6594e2a6dd7c3dc745eb856025276ec7d1f Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Wed, 11 Feb 2015 15:53:56 +0000 Subject: First step of making user_rooms_intersect() faster - implement in intersection logic in Python code terms of a DB query that is cacheable per user --- synapse/storage/roommember.py | 38 ++++++++++++++++++++------------------ 1 file changed, 20 insertions(+), 18 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index c69dd995ce..d490a374e5 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -240,28 +240,30 @@ class RoomMemberStore(SQLBaseStore): results = self._parse_events_txn(txn, rows) return results + @defer.inlineCallbacks def user_rooms_intersect(self, user_id_list): """ Checks whether all the users whose IDs are given in a list share a room. + + This is a "hot path" function that's called a lot, e.g. by presence for + generating the event stream. """ - def interaction(txn): - user_list_clause = " OR ".join(["m.user_id = ?"] * len(user_id_list)) - sql = ( - "SELECT m.room_id FROM room_memberships as m " - "INNER JOIN current_state_events as c " - "ON m.event_id = c.event_id " - "WHERE m.membership = 'join' " - "AND (%(clause)s) " - # TODO(paul): We've got duplicate rows in the database somewhere - # so we have to DISTINCT m.user_id here - "GROUP BY m.room_id HAVING COUNT(DISTINCT m.user_id) = ?" - ) % {"clause": user_list_clause} - - args = list(user_id_list) - args.append(len(user_id_list)) + if len(user_id_list) < 2: + defer.returnValue(True) - txn.execute(sql, args) + deferreds = [ + self.get_rooms_for_user_where_membership_is( + u, membership_list=[Membership.JOIN], + ) + for u in user_id_list + ] + + results = yield defer.DeferredList(deferreds) + + # A list of sets of strings giving room IDs for each user + room_id_lists = [set([r.room_id for r in result[1]]) for result in results] - return len(txn.fetchall()) > 0 + # There isn't a setintersection(*list_of_sets) + ret = len(room_id_lists.pop(0).intersection(*room_id_lists)) > 0 - return self.runInteraction("user_rooms_intersect", interaction) + defer.returnValue(ret) -- cgit 1.4.1 From 45b56609ae84e7ffc3713335e7d9abc315ad1725 Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Wed, 11 Feb 2015 16:04:08 +0000 Subject: Cache the result of a get_rooms_for_user query, to make user_rooms_intersect() much lighter in the read-common case --- synapse/storage/roommember.py | 41 ++++++++++++++++++++++++++++++++++------- 1 file changed, 34 insertions(+), 7 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index d490a374e5..e05465bc13 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -35,6 +35,11 @@ RoomsForUser = namedtuple( class RoomMemberStore(SQLBaseStore): + def __init__(self, *args, **kw): + super(RoomMemberStore, self).__init__(*args, **kw) + + self._user_rooms_cache = {} + def _store_room_member_txn(self, txn, event): """Store a room member in the database. """ @@ -98,6 +103,8 @@ class RoomMemberStore(SQLBaseStore): txn.execute(sql, (event.room_id, domain)) + self.invalidate_rooms_for_user(target_user_id) + @defer.inlineCallbacks def get_room_member(self, user_id, room_id): """Retrieve the current state of a room member. @@ -240,23 +247,43 @@ class RoomMemberStore(SQLBaseStore): results = self._parse_events_txn(txn, rows) return results + # TODO(paul): Create a nice @cached decorator to do this + # @cached + # def get_foo(...) + # ... + # invalidate_foo = get_foo.invalidator + + @defer.inlineCallbacks + def get_rooms_for_user(self, user_id): + # TODO(paul): put some performance counters in here so we can easily + # track what impact this cache is having + if user_id in self._user_rooms_cache: + defer.returnValue(self._user_rooms_cache[user_id]) + + rooms = yield self.get_rooms_for_user_where_membership_is( + user_id, membership_list=[Membership.JOIN], + ) + + self._user_rooms_cache[user_id] = rooms + defer.returnValue(rooms) + + def invalidate_rooms_for_user(self, user_id): + if user_id in self._user_rooms_cache: + del self._user_rooms_cache[user_id] + @defer.inlineCallbacks def user_rooms_intersect(self, user_id_list): """ Checks whether all the users whose IDs are given in a list share a room. This is a "hot path" function that's called a lot, e.g. by presence for - generating the event stream. + generating the event stream. As such, it is implemented locally by + wrapping logic around heavily-cached database queries. """ if len(user_id_list) < 2: defer.returnValue(True) - deferreds = [ - self.get_rooms_for_user_where_membership_is( - u, membership_list=[Membership.JOIN], - ) - for u in user_id_list - ] + deferreds = [self.get_rooms_for_user(u) for u in user_id_list] results = yield defer.DeferredList(deferreds) -- cgit 1.4.1 From 7f47ba7383302fdbdaa3a10abef00d3710c77fce Mon Sep 17 00:00:00 2001 From: "Paul \"LeoNerd\" Evans" Date: Wed, 11 Feb 2015 16:18:21 +0000 Subject: Added another TODO note --- synapse/storage/roommember.py | 3 +++ 1 file changed, 3 insertions(+) (limited to 'synapse/storage') diff --git a/synapse/storage/roommember.py b/synapse/storage/roommember.py index e05465bc13..779f9ce544 100644 --- a/synapse/storage/roommember.py +++ b/synapse/storage/roommember.py @@ -264,6 +264,9 @@ class RoomMemberStore(SQLBaseStore): user_id, membership_list=[Membership.JOIN], ) + # TODO(paul): Consider applying a maximum size; just evict things at + # random, or consider LRU? + self._user_rooms_cache[user_id] = rooms defer.returnValue(rooms) -- cgit 1.4.1