From 755def8083ec887feabcb45b3bc111db4aef20ab Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Mon, 18 May 2015 13:46:47 +0100 Subject: Add more doc string, reduce C+P boilerplate for getting room list --- synapse/handlers/presence.py | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index a01020e202..ce9dd64394 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -297,7 +297,26 @@ class PresenceHandler(BaseHandler): self.changed_presencelike_data(user, {"last_active": now}) + def get_joined_rooms_for_user(self, user): + """Get the list of rooms a user is joined to. + + Args: + user(UserID): The user. + Returns: + A Deferred of a list of room id strings. + """ + rm_handler = self.homeserver.get_handlers().room_member_handler + return rm_handler.get_joined_rooms_for_user(user) + def changed_presencelike_data(self, user, state): + """Updates the presence state of a local user. + + Args: + user(UserID): The user being updated. + state(dict): The new presence state for the user. + Returns: + A Deferred + """ statuscache = self._get_or_make_usercache(user) self._user_cachemap_latest_serial += 1 @@ -544,8 +563,7 @@ class PresenceHandler(BaseHandler): # Also include people in all my rooms - rm_handler = self.homeserver.get_handlers().room_member_handler - room_ids = yield rm_handler.get_joined_rooms_for_user(user) + room_ids = yield self.get_joined_rooms_for_user(user) if state is None: state = yield self.store.get_presence_state(user.localpart) @@ -745,8 +763,7 @@ class PresenceHandler(BaseHandler): # and also user is informed of server-forced pushes localusers.add(user) - rm_handler = self.homeserver.get_handlers().room_member_handler - room_ids = yield rm_handler.get_joined_rooms_for_user(user) + room_ids = yield self.get_joined_rooms_for_user(user) if not localusers and not room_ids: defer.returnValue(None) @@ -791,8 +808,7 @@ class PresenceHandler(BaseHandler): " | %d interested local observers %r", len(observers), observers ) - rm_handler = self.homeserver.get_handlers().room_member_handler - room_ids = yield rm_handler.get_joined_rooms_for_user(user) + room_ids = yield self.get_joined_rooms_for_user(user) if room_ids: logger.debug(" | %d interested room IDs %r", len(room_ids), room_ids) -- cgit 1.4.1 From e1150cac4bceab88097ea2421323f3b3852028e3 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Mon, 18 May 2015 15:46:37 +0100 Subject: Move updating the serial and state of the presence cache into a single function --- synapse/handlers/presence.py | 60 +++++++++++++++++++++++++++++++------------- 1 file changed, 43 insertions(+), 17 deletions(-) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 0c3290b307..d129d4ca8a 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -308,6 +308,11 @@ class PresenceHandler(BaseHandler): rm_handler = self.homeserver.get_handlers().room_member_handler return rm_handler.get_joined_rooms_for_user(user) + def get_joined_users_for_room_id(self, room_id): + rm_handler = self.homeserver.get_handlers().room_member_handler + return rm_handler.get_room_members(room_id) + + @defer.inlineCallbacks def changed_presencelike_data(self, user, state): """Updates the presence state of a local user. @@ -317,12 +322,9 @@ class PresenceHandler(BaseHandler): Returns: A Deferred """ - statuscache = self._get_or_make_usercache(user) - self._user_cachemap_latest_serial += 1 - statuscache.update(state, serial=self._user_cachemap_latest_serial) - - return self.push_presence(user, statuscache=statuscache) + statuscache = yield self.update_presence_cache(user, state) + yield self.push_presence(user, statuscache=statuscache) @log_function def started_user_eventstream(self, user): @@ -345,13 +347,12 @@ class PresenceHandler(BaseHandler): room_id(str): The room id the user joined. """ if self.hs.is_mine(user): - statuscache = self._get_or_make_usercache(user) - # No actual update but we need to bump the serial anyway for the # event source self._user_cachemap_latest_serial += 1 - statuscache.update({}, serial=self._user_cachemap_latest_serial) - + statuscache = yield self.update_presence_cache( + user, room_ids=[room_id] + ) self.push_update_to_local_and_remote( observed_user=user, room_ids=[room_id], @@ -359,16 +360,17 @@ class PresenceHandler(BaseHandler): ) # We also want to tell them about current presence of people. - rm_handler = self.homeserver.get_handlers().room_member_handler - curr_users = yield rm_handler.get_room_members(room_id) + curr_users = yield self.get_joined_users_for_room_id(room_id) for local_user in [c for c in curr_users if self.hs.is_mine(c)]: - statuscache = self._get_or_offline_usercache(local_user) - statuscache.update({}, serial=self._user_cachemap_latest_serial) + statuscache = yield self.update_presence_cache( + local_user, room_ids=[room_id], add_to_cache=False + ) + self.push_update_to_local_and_remote( observed_user=local_user, users_to_push=[user], - statuscache=self._get_or_offline_usercache(local_user), + statuscache=statuscache, ) @defer.inlineCallbacks @@ -829,10 +831,8 @@ class PresenceHandler(BaseHandler): self.clock.time_msec() - state.pop("last_active_ago") ) - statuscache = self._get_or_make_usercache(user) - self._user_cachemap_latest_serial += 1 - statuscache.update(state, serial=self._user_cachemap_latest_serial) + yield self.update_presence_cache(user, state, room_ids=room_ids) if not observers and not room_ids: logger.debug(" | no interested observers or room IDs") @@ -890,6 +890,32 @@ class PresenceHandler(BaseHandler): yield defer.DeferredList(deferreds, consumeErrors=True) + @defer.inlineCallbacks + def update_presence_cache(self, user, state={}, room_ids=None, + add_to_cache=True): + """Update the presence cache for a user with a new state and bump the + serial to the latest value. + + Args: + user(UserID): The user being updated + state(dict): The presence state being updated + room_ids(None or list of str): A list of room_ids to update. If + room_ids is None then fetch the list of room_ids the user is + joined to. + add_to_cache: Whether to add an entry to the presence cache if the + user isn't already in the cache. + Returns: + A Deferred UserPresenceCache for the user being updated. + """ + if room_ids is None: + room_ids = yield self.get_joined_rooms_for_user(user) + if add_to_cache: + statuscache = self._get_or_make_usercache(user) + else: + statuscache = self._get_or_offline_usercache(user) + statuscache.update(state, serial=self._user_cachemap_latest_serial) + defer.returnValue(statuscache) + @defer.inlineCallbacks def push_update_to_local_and_remote(self, observed_user, statuscache, users_to_push=[], room_ids=[], -- cgit 1.4.1 From 591c4bf223a4a8698f51ba258984e769f593e32b Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Mon, 18 May 2015 16:21:51 +0100 Subject: Cache the most recent serial for each room --- synapse/handlers/presence.py | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index d129d4ca8a..aa1d73f2f9 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -146,6 +146,10 @@ class PresenceHandler(BaseHandler): self._user_cachemap = {} self._user_cachemap_latest_serial = 0 + # map room_ids to the latest presence serial for a member of that + # room + self._room_serials = {} + metrics.register_callback( "userCachemap:size", lambda: len(self._user_cachemap), @@ -909,6 +913,9 @@ class PresenceHandler(BaseHandler): """ if room_ids is None: room_ids = yield self.get_joined_rooms_for_user(user) + + for room_id in room_ids: + self._room_serials[room_id] = self._user_cachemap_latest_serial if add_to_cache: statuscache = self._get_or_make_usercache(user) else: @@ -1069,8 +1076,6 @@ class PresenceEventSource(object): def get_new_events_for_user(self, user, from_key, limit): from_key = int(from_key) - observer_user = user - presence = self.hs.get_handlers().presence_handler cachemap = presence._user_cachemap @@ -1079,17 +1084,28 @@ class PresenceEventSource(object): clock = self.clock latest_serial = 0 + presence_list = yield presence.store.get_presence_list( + user.localpart, accepted=True + ) + if presence_list is None: + presence_list = () + user_ids_to_check = set( + UserID.from_string(p["observed_user_id"]) for p in presence_list + ) + room_ids = yield presence.get_joined_rooms_for_user(user) + for room_id in set(room_ids) & set(presence._room_serials): + if presence._room_serials[room_id] > from_key: + joined = yield presence.get_joined_users_for_room_id(room_id) + user_ids_to_check |= set(joined) + updates = [] # TODO(paul): use a DeferredList ? How to limit concurrency. - for observed_user in cachemap.keys(): + for observed_user in user_ids_to_check & set(cachemap): cached = cachemap[observed_user] if cached.serial <= from_key or cached.serial > max_serial: continue - if not (yield self.is_visible(observer_user, observed_user)): - continue - latest_serial = max(cached.serial, latest_serial) updates.append(cached.make_event(user=observed_user, clock=clock)) -- cgit 1.4.1 From e4c65b338d44fadc058cbd8e4cd79ae1601d3526 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Mon, 18 May 2015 18:21:06 +0100 Subject: Speed up the get_pagination_rows as well --- synapse/handlers/presence.py | 20 +++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index aa1d73f2f9..6537a37385 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -1154,14 +1154,28 @@ class PresenceEventSource(object): presence = self.hs.get_handlers().presence_handler cachemap = presence._user_cachemap + user_ids_to_check = {user} + presence_list = yield presence.store.get_presence_list( + user.localpart, accepted=True + ) + if presence_list is None: + presence_list = () + user_ids_to_check |= set( + UserID.from_string(p["observed_user_id"]) for p in presence_list + ) + room_ids = yield presence.get_joined_rooms_for_user(user) + for room_id in set(room_ids) & set(presence._room_serials): + if presence._room_serials[room_id] >= from_key: + joined = yield presence.get_joined_users_for_room_id(room_id) + user_ids_to_check |= set(joined) + updates = [] # TODO(paul): use a DeferredList ? How to limit concurrency. - for observed_user in cachemap.keys(): + for observed_user in user_ids_to_check & set(cachemap): if not (to_key < cachemap[observed_user].serial <= from_key): continue - if (yield self.is_visible(observer_user, observed_user)): - updates.append((observed_user, cachemap[observed_user])) + updates.append((observed_user, cachemap[observed_user])) # TODO(paul): limit -- cgit 1.4.1 From e01b825cc929e16b6a60be0688bbe6d8d9b3866e Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 20 May 2015 13:21:59 +0100 Subject: Clean up the presence_list checking logic a bit --- synapse/handlers/presence.py | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 6537a37385..226d6a0f51 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -1084,14 +1084,14 @@ class PresenceEventSource(object): clock = self.clock latest_serial = 0 + user_ids_to_check = {user} presence_list = yield presence.store.get_presence_list( user.localpart, accepted=True ) - if presence_list is None: - presence_list = () - user_ids_to_check = set( - UserID.from_string(p["observed_user_id"]) for p in presence_list - ) + if presence_list is not None: + user_ids_to_check |= set( + UserID.from_string(p["observed_user_id"]) for p in presence_list + ) room_ids = yield presence.get_joined_rooms_for_user(user) for room_id in set(room_ids) & set(presence._room_serials): if presence._room_serials[room_id] > from_key: @@ -1142,8 +1142,6 @@ class PresenceEventSource(object): def get_pagination_rows(self, user, pagination_config, key): # TODO (erikj): Does this make sense? Ordering? - observer_user = user - from_key = int(pagination_config.from_key) if pagination_config.to_key: @@ -1158,11 +1156,10 @@ class PresenceEventSource(object): presence_list = yield presence.store.get_presence_list( user.localpart, accepted=True ) - if presence_list is None: - presence_list = () - user_ids_to_check |= set( - UserID.from_string(p["observed_user_id"]) for p in presence_list - ) + if presence_list is not None: + user_ids_to_check |= set( + UserID.from_string(p["observed_user_id"]) for p in presence_list + ) room_ids = yield presence.get_joined_rooms_for_user(user) for room_id in set(room_ids) & set(presence._room_serials): if presence._room_serials[room_id] >= from_key: -- cgit 1.4.1 From 8eca5bd50abc9deb3cd428f3f6b3b8fbeb8bdee1 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 20 May 2015 13:22:18 +0100 Subject: Fix the presence tests --- tests/handlers/test_presence.py | 13 +++---------- tests/rest/client/v1/test_presence.py | 3 +++ 2 files changed, 6 insertions(+), 10 deletions(-) diff --git a/tests/handlers/test_presence.py b/tests/handlers/test_presence.py index ee773797e7..12cf5747a2 100644 --- a/tests/handlers/test_presence.py +++ b/tests/handlers/test_presence.py @@ -624,6 +624,7 @@ class PresencePushTestCase(MockedDatastorePresenceTestCase): """ PRESENCE_LIST = { 'apple': [ "@banana:test", "@clementine:test" ], + 'banana': [ "@apple:test" ], } @defer.inlineCallbacks @@ -836,12 +837,7 @@ class PresencePushTestCase(MockedDatastorePresenceTestCase): @defer.inlineCallbacks def test_recv_remote(self): - # TODO(paul): Gut-wrenching - potato_set = self.handler._remote_recvmap.setdefault(self.u_potato, - set()) - potato_set.add(self.u_apple) - - self.room_members = [self.u_banana, self.u_potato] + self.room_members = [self.u_apple, self.u_banana, self.u_potato] self.assertEquals(self.event_source.get_current_key(), 0) @@ -886,11 +882,8 @@ class PresencePushTestCase(MockedDatastorePresenceTestCase): @defer.inlineCallbacks def test_recv_remote_offline(self): """ Various tests relating to SYN-261 """ - potato_set = self.handler._remote_recvmap.setdefault(self.u_potato, - set()) - potato_set.add(self.u_apple) - self.room_members = [self.u_banana, self.u_potato] + self.room_members = [self.u_apple, self.u_banana, self.u_potato] self.assertEquals(self.event_source.get_current_key(), 0) diff --git a/tests/rest/client/v1/test_presence.py b/tests/rest/client/v1/test_presence.py index 29c0038f06..8f3df92418 100644 --- a/tests/rest/client/v1/test_presence.py +++ b/tests/rest/client/v1/test_presence.py @@ -295,6 +295,9 @@ class PresenceEventStreamTestCase(unittest.TestCase): else: return [] hs.handlers.room_member_handler.get_joined_rooms_for_user = get_rooms_for_user + hs.handlers.room_member_handler.get_room_members = ( + lambda r: self.room_members if r == "a-room" else [] + ) self.mock_datastore = hs.get_datastore() self.mock_datastore.get_app_service_by_token = Mock(return_value=None) -- cgit 1.4.1 From 7ae8afb7ef5a0fb3162339737682e9248980600d Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Wed, 20 May 2015 14:48:11 +0100 Subject: Removed unused 'is_visible' method --- synapse/handlers/presence.py | 27 --------------------------- 1 file changed, 27 deletions(-) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 226d6a0f51..6c48b1d20e 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -1045,32 +1045,6 @@ class PresenceEventSource(object): self.hs = hs self.clock = hs.get_clock() - @defer.inlineCallbacks - def is_visible(self, observer_user, observed_user): - if observer_user == observed_user: - defer.returnValue(True) - - presence = self.hs.get_handlers().presence_handler - - if (yield presence.store.user_rooms_intersect( - [u.to_string() for u in observer_user, observed_user])): - defer.returnValue(True) - - if self.hs.is_mine(observed_user): - pushmap = presence._local_pushmap - - defer.returnValue( - observed_user.localpart in pushmap and - observer_user in pushmap[observed_user.localpart] - ) - else: - recvmap = presence._remote_recvmap - - defer.returnValue( - observed_user in recvmap and - observer_user in recvmap[observed_user] - ) - @defer.inlineCallbacks @log_function def get_new_events_for_user(self, user, from_key, limit): @@ -1099,7 +1073,6 @@ class PresenceEventSource(object): user_ids_to_check |= set(joined) updates = [] - # TODO(paul): use a DeferredList ? How to limit concurrency. for observed_user in user_ids_to_check & set(cachemap): cached = cachemap[observed_user] -- cgit 1.4.1 From 106a3051b88be742d24ace05f72d9ab6bff29dd2 Mon Sep 17 00:00:00 2001 From: Mark Haines Date: Fri, 22 May 2015 15:53:03 +0100 Subject: Remove spurious TODO comment --- synapse/handlers/presence.py | 1 - 1 file changed, 1 deletion(-) diff --git a/synapse/handlers/presence.py b/synapse/handlers/presence.py index 6c48b1d20e..670c1d353f 100644 --- a/synapse/handlers/presence.py +++ b/synapse/handlers/presence.py @@ -1140,7 +1140,6 @@ class PresenceEventSource(object): user_ids_to_check |= set(joined) updates = [] - # TODO(paul): use a DeferredList ? How to limit concurrency. for observed_user in user_ids_to_check & set(cachemap): if not (to_key < cachemap[observed_user].serial <= from_key): continue -- cgit 1.4.1