From 447aed42d22d3ece245c69f397d348b3a5b7bfa8 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Sat, 27 Jan 2018 15:40:41 +0000 Subject: Add event_map param to resolve_state_groups --- synapse/storage/events.py | 1 + 1 file changed, 1 insertion(+) (limited to 'synapse/storage') diff --git a/synapse/storage/events.py b/synapse/storage/events.py index 2fead9eb0f..7b912ad413 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -586,6 +586,7 @@ class EventsStore(SQLBaseStore): current_state = yield resolve_events_with_factory( state_sets, + event_map={}, state_map_factory=get_events, ) defer.returnValue(current_state) -- cgit 1.4.1 From 9fcbbe8e7d7557fef7fe03533166b376d6fa82ef Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Sat, 27 Jan 2018 09:49:15 +0000 Subject: Check that events being persisted have state_group --- synapse/storage/events.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/storage/events.py b/synapse/storage/events.py index 7b912ad413..9bceded7ba 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -515,16 +515,21 @@ class EventsStore(SQLBaseStore): if ctx.current_state_ids is None: raise Exception("Unknown current state") + if ctx.state_group is None: + # I don't think this can happen, but let's double-check + raise Exception( + "Context for new extremity event %s has no state " + "group" % event_id, + ) + # If we've already seen the state group don't bother adding # it to the state sets again if ctx.state_group not in state_groups: state_sets.append(ctx.current_state_ids) if ctx.delta_ids or hasattr(ev, "state_key"): was_updated = True - if ctx.state_group: - # Add this as a seen state group (if it has a state - # group) - state_groups.add(ctx.state_group) + # Add this as a seen state group + state_groups.add(ctx.state_group) break else: # If we couldn't find it, then we'll need to pull -- cgit 1.4.1 From 225dc3b4cb8875fff52180d2f3b1e386dec26f4d Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Tue, 30 Jan 2018 10:17:55 +0000 Subject: Flatten _get_new_state_after_events rejig the if statements to simplify the logic and reduce indentation --- synapse/storage/events.py | 90 ++++++++++++++++++++++++----------------------- 1 file changed, 46 insertions(+), 44 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/storage/events.py b/synapse/storage/events.py index 9bceded7ba..1b5dffe1c7 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -503,6 +503,10 @@ class EventsStore(SQLBaseStore): None if there are no changes to the room state, or a dict of (type, state_key) -> event_id]. """ + + if not new_latest_event_ids: + defer.returnValue({}) + state_sets = [] state_groups = set() missing_event_ids = [] @@ -537,6 +541,9 @@ class EventsStore(SQLBaseStore): was_updated = True missing_event_ids.append(event_id) + if not was_updated: + return + if missing_event_ids: # Now pull out the state for any missing events from DB event_to_groups = yield self._get_state_group_for_events( @@ -549,54 +556,49 @@ class EventsStore(SQLBaseStore): group_to_state = yield self._get_state_for_groups(groups) state_sets.extend(group_to_state.itervalues()) - if not new_latest_event_ids: - defer.returnValue({}) - elif was_updated: - if len(state_sets) == 1: - # If there is only one state set, then we know what the current - # state is. - defer.returnValue(state_sets[0]) - else: - # We work out the current state by passing the state sets to the - # state resolution algorithm. It may ask for some events, including - # the events we have yet to persist, so we need a slightly more - # complicated event lookup function than simply looking the events - # up in the db. - - logger.info( - "Resolving state with %i state sets", len(state_sets), - ) + if len(state_sets) == 1: + # If there is only one state set, then we know what the current + # state is. + defer.returnValue(state_sets[0]) - events_map = {ev.event_id: ev for ev, _ in events_context} - - @defer.inlineCallbacks - def get_events(ev_ids): - # We get the events by first looking at the list of events we - # are trying to persist, and then fetching the rest from the DB. - db = [] - to_return = {} - for ev_id in ev_ids: - ev = events_map.get(ev_id, None) - if ev: - to_return[ev_id] = ev - else: - db.append(ev_id) + # We work out the current state by passing the state sets to the + # state resolution algorithm. It may ask for some events, including + # the events we have yet to persist, so we need a slightly more + # complicated event lookup function than simply looking the events + # up in the db. - if db: - evs = yield self.get_events( - ev_ids, get_prev_content=False, check_redacted=False, - ) - to_return.update(evs) - defer.returnValue(to_return) + logger.info( + "Resolving state with %i state sets", len(state_sets), + ) - current_state = yield resolve_events_with_factory( - state_sets, - event_map={}, - state_map_factory=get_events, + events_map = {ev.event_id: ev for ev, _ in events_context} + + @defer.inlineCallbacks + def get_events(ev_ids): + # We get the events by first looking at the list of events we + # are trying to persist, and then fetching the rest from the DB. + db = [] + to_return = {} + for ev_id in ev_ids: + ev = events_map.get(ev_id, None) + if ev: + to_return[ev_id] = ev + else: + db.append(ev_id) + + if db: + evs = yield self.get_events( + ev_ids, get_prev_content=False, check_redacted=False, ) - defer.returnValue(current_state) - else: - return + to_return.update(evs) + defer.returnValue(to_return) + + current_state = yield resolve_events_with_factory( + state_sets, + event_map={}, + state_map_factory=get_events, + ) + defer.returnValue(current_state) @defer.inlineCallbacks def _calculate_state_delta(self, room_id, current_state): -- cgit 1.4.1 From ebfe64e3d69d0047ee9902a05beaf0249f11e072 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Tue, 30 Jan 2018 11:06:15 +0000 Subject: Use StateResolutionHandler to resolve state in persist events ... and thus benefit (hopefully) from its cache. --- synapse/storage/events.py | 72 ++++++++++++++++------------------------------- 1 file changed, 24 insertions(+), 48 deletions(-) (limited to 'synapse/storage') diff --git a/synapse/storage/events.py b/synapse/storage/events.py index 1b5dffe1c7..ca1d4a3986 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -27,7 +27,6 @@ from synapse.util.logutils import log_function from synapse.util.metrics import Measure from synapse.api.constants import EventTypes from synapse.api.errors import SynapseError -from synapse.state import resolve_events_with_factory from synapse.util.caches.descriptors import cached from synapse.types import get_domain_from_id @@ -237,6 +236,8 @@ class EventsStore(SQLBaseStore): self._event_persist_queue = _EventPeristenceQueue() + self._state_resolution_handler = hs.get_state_resolution_handler() + def persist_events(self, events_and_contexts, backfilled=False): """ Write events to the database @@ -402,6 +403,7 @@ class EventsStore(SQLBaseStore): "Calculating state delta for room %s", room_id, ) current_state = yield self._get_new_state_after_events( + room_id, ev_ctx_rm, new_latest_event_ids, ) if current_state is not None: @@ -487,11 +489,14 @@ class EventsStore(SQLBaseStore): defer.returnValue(new_latest_event_ids) @defer.inlineCallbacks - def _get_new_state_after_events(self, events_context, new_latest_event_ids): + def _get_new_state_after_events(self, room_id, events_context, new_latest_event_ids): """Calculate the current state dict after adding some new events to a room Args: + room_id (str): + room to which the events are being added. Used for logging etc + events_context (list[(EventBase, EventContext)]): events and contexts which are being added to the room @@ -507,8 +512,8 @@ class EventsStore(SQLBaseStore): if not new_latest_event_ids: defer.returnValue({}) - state_sets = [] - state_groups = set() + # map from state_group to ((type, key) -> event_id) state map + state_groups = {} missing_event_ids = [] was_updated = False for event_id in new_latest_event_ids: @@ -529,11 +534,9 @@ class EventsStore(SQLBaseStore): # If we've already seen the state group don't bother adding # it to the state sets again if ctx.state_group not in state_groups: - state_sets.append(ctx.current_state_ids) + state_groups[ctx.state_group] = ctx.current_state_ids if ctx.delta_ids or hasattr(ev, "state_key"): was_updated = True - # Add this as a seen state group - state_groups.add(ctx.state_group) break else: # If we couldn't find it, then we'll need to pull @@ -550,55 +553,28 @@ class EventsStore(SQLBaseStore): missing_event_ids, ) - groups = set(event_to_groups.itervalues()) - state_groups + groups = set(event_to_groups.itervalues()) - set(state_groups.iterkeys()) if groups: group_to_state = yield self._get_state_for_groups(groups) - state_sets.extend(group_to_state.itervalues()) + state_groups.update(group_to_state) - if len(state_sets) == 1: - # If there is only one state set, then we know what the current + if len(state_groups) == 1: + # If there is only one state group, then we know what the current # state is. - defer.returnValue(state_sets[0]) - - # We work out the current state by passing the state sets to the - # state resolution algorithm. It may ask for some events, including - # the events we have yet to persist, so we need a slightly more - # complicated event lookup function than simply looking the events - # up in the db. - - logger.info( - "Resolving state with %i state sets", len(state_sets), - ) - - events_map = {ev.event_id: ev for ev, _ in events_context} + defer.returnValue(state_groups.values()[0]) - @defer.inlineCallbacks def get_events(ev_ids): - # We get the events by first looking at the list of events we - # are trying to persist, and then fetching the rest from the DB. - db = [] - to_return = {} - for ev_id in ev_ids: - ev = events_map.get(ev_id, None) - if ev: - to_return[ev_id] = ev - else: - db.append(ev_id) - - if db: - evs = yield self.get_events( - ev_ids, get_prev_content=False, check_redacted=False, - ) - to_return.update(evs) - defer.returnValue(to_return) - - current_state = yield resolve_events_with_factory( - state_sets, - event_map={}, - state_map_factory=get_events, + return self.get_events( + ev_ids, get_prev_content=False, check_redacted=False, + ) + events_map = {ev.event_id: ev for ev, _ in events_context} + logger.debug("calling resolve_state_groups from preserve_events") + res = yield self._state_resolution_handler.resolve_state_groups( + room_id, state_groups, events_map, get_events ) - defer.returnValue(current_state) + + defer.returnValue(res.state) @defer.inlineCallbacks def _calculate_state_delta(self, room_id, current_state): -- cgit 1.4.1 From 630caf8a703250e0f568000958faee42f9336b72 Mon Sep 17 00:00:00 2001 From: Richard van der Hoff Date: Tue, 13 Feb 2018 14:29:22 +0000 Subject: style nit --- synapse/storage/events.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'synapse/storage') diff --git a/synapse/storage/events.py b/synapse/storage/events.py index ca1d4a3986..3d5eb9bc02 100644 --- a/synapse/storage/events.py +++ b/synapse/storage/events.py @@ -528,7 +528,7 @@ class EventsStore(SQLBaseStore): # I don't think this can happen, but let's double-check raise Exception( "Context for new extremity event %s has no state " - "group" % event_id, + "group" % (event_id, ), ) # If we've already seen the state group don't bother adding -- cgit 1.4.1