synapse/state.py
def resolve_state_groups(
        self, room_id, state_groups_ids, event_map, state_map_factory,
    ):
        """Resolves conflicts between a set of state groups

        Always generates a new state group (unless we hit the cache), so should
        not be called for a single state group

        Args:
            room_id (str): room we are resolving for (used for logging)
            state_groups_ids (dict[int, dict[(str, str), str]]):
                 map from state group id to the state in that state group
                (where 'state' is a map from state key to event id)

            event_map(dict[str,FrozenEvent]|None):
                a dict from event_id to event, for any events that we happen to
                have in flight (eg, those currently being persisted). This will be
                used as a starting point fof finding the state we need; any missing
                events will be requested via state_map_factory.

                If None, all events will be fetched via state_map_factory.

        Returns:
            Deferred[_StateCacheEntry]: resolved state
        """
        logger.debug(
            "resolve_state_groups state_groups %s",
            state_groups_ids.keys()
        )

        group_names = frozenset(state_groups_ids.keys())

        with (yield self.resolve_linearizer.queue(group_names)):
            if self._state_cache is not None:
                cache = self._state_cache.get(group_names, None)
                if cache:
                    defer.returnValue(cache)

            logger.info(
                "Resolving state for %s with %d groups", room_id, len(state_groups_ids)
            )

            # build a map from state key to the event_ids which set that state.
            # dict[(str, str), set[str])
            state = {}
            for st in state_groups_ids.itervalues():
                for key, e_id in st.iteritems():
                    state.setdefault(key, set()).add(e_id)

            # build a map from state key to the event_ids which set that state,
            # including only those where there are state keys in conflict.
            conflicted_state = {
                k: list(v)
                for k, v in state.iteritems()
                if len(v) > 1
            }

            if conflicted_state:
                logger.info("Resolving conflicted state for %r", room_id)
                with Measure(self.clock, "state._resolve_events"):
                    new_state = yield resolve_events_with_factory(
                        state_groups_ids.values(),
                        event_map=event_map,
                        state_map_factory=state_map_factory,
                    )
            else:
                new_state = {
                    key: e_ids.pop() for key, e_ids in state.iteritems()
                }

            with Measure(self.clock, "state.create_group_ids"):
                # if the new state matches any of the input state groups, we can
                # use that state group again. Otherwise we will generate a state_id
                # which will be used as a cache key for future resolutions, but
                # not get persisted.
                state_group = None
                new_state_event_ids = frozenset(new_state.itervalues())
                for sg, events in state_groups_ids.iteritems():
                    if new_state_event_ids == frozenset(e_id for e_id in events):
                        state_group = sg
                        break

                # TODO: We want to create a state group for this set of events, to
                # increase cache hits, but we need to make sure that it doesn't
                # end up as a prev_group without being added to the database

                prev_group = None
                delta_ids = None
                for old_group, old_ids in state_groups_ids.iteritems():
                    if not set(new_state) - set(old_ids):
                        n_delta_ids = {
                            k: v
                            for k, v in new_state.iteritems()
                            if old_ids.get(k) != v
                        }
                        if not delta_ids or len(n_delta_ids) < len(delta_ids):
                            prev_group = old_group
                            delta_ids = n_delta_ids

            cache = _StateCacheEntry(
                state=new_state,
                state_group=state_group,
                prev_group=prev_group,
                delta_ids=delta_ids,
            )

            if self._state_cache is not None:
                self._state_cache[group_names] = cache

            defer.returnValue(cache)
Resolves conflicts between a set of state groups

        Always generates a new state group (unless we hit the cache), so should
        not be called for a single state group

        Args:
            room_id (str): room we are resolving for (used for logging)
            state_groups_ids (dict[int, dict[(str, str), str]]):
                 map from state group id to the state in that state group
                (where 'state' is a map from state key to event id)

            event_map(dict[str,FrozenEvent]|None):
                a dict from event_id to event, for any events that we happen to
                have in flight (eg, those currently being persisted). This will be
                used as a starting point fof finding the state we need; any missing
                events will be requested via state_map_factory.

                If None, all events will be fetched via state_map_factory.

        Returns:
            Deferred[_StateCacheEntry]: resolved state
        
Resolves conflicts between a set of state groups
Always generates a new state group (unless we hit the cache), so should
not be called for a single state group
Resolves conflicts between a set of state groups Always generates a new state group ( unless we hit the cache ) so should not be called for a single state group
def resolve_state_groups(
        self, room_id, state_groups_ids, event_map, state_map_factory,
    ):
        logger.debug(
            "resolve_state_groups state_groups %s",
            state_groups_ids.keys()
        )
        group_names = frozenset(state_groups_ids.keys())
        with (yield self.resolve_linearizer.queue(group_names)):
            if self._state_cache is not None:
                cache = self._state_cache.get(group_names, None)
                if cache:
                    defer.returnValue(cache)
            logger.info(
                "Resolving state for %s with %d groups", room_id, len(state_groups_ids)
            )
            state = {}
            for st in state_groups_ids.itervalues():
                for key, e_id in st.iteritems():
                    state.setdefault(key, set()).add(e_id)
            conflicted_state = {
                k: list(v)
                for k, v in state.iteritems()
                if len(v) > 1
            }
            if conflicted_state:
                logger.info("Resolving conflicted state for %r", room_id)
                with Measure(self.clock, "state._resolve_events"):
                    new_state = yield resolve_events_with_factory(
                        state_groups_ids.values(),
                        event_map=event_map,
                        state_map_factory=state_map_factory,
                    )
            else:
                new_state = {
                    key: e_ids.pop() for key, e_ids in state.iteritems()
                }
            with Measure(self.clock, "state.create_group_ids"):
                state_group = None
                new_state_event_ids = frozenset(new_state.itervalues())
                for sg, events in state_groups_ids.iteritems():
                    if new_state_event_ids == frozenset(e_id for e_id in events):
                        state_group = sg
                        break
                prev_group = None
                delta_ids = None
                for old_group, old_ids in state_groups_ids.iteritems():
                    if not set(new_state) - set(old_ids):
                        n_delta_ids = {
                            k: v
                            for k, v in new_state.iteritems()
                            if old_ids.get(k) != v
                        }
                        if not delta_ids or len(n_delta_ids) < len(delta_ids):
                            prev_group = old_group
                            delta_ids = n_delta_ids
            cache = _StateCacheEntry(
                state=new_state,
                state_group=state_group,
                prev_group=prev_group,
                delta_ids=delta_ids,
            )
            if self._state_cache is not None:
                self._state_cache[group_names] = cache
            defer.returnValue(cache)
def resolve_state_groups ( self , room_id , state_groups_ids , event_map , state_map_factory , ) : logger . debug ( "resolve_state_groups state_groups %s" , state_groups_ids . keys ( ) ) group_names = frozenset ( state_groups_ids . keys ( ) ) with ( yield self . resolve_linearizer . queue ( group_names ) ) : if self . _state_cache is not None : cache = self . _state_cache . get ( group_names , None ) if cache : defer . returnValue ( cache ) logger . info ( "Resolving state for %s with %d groups" , room_id , len ( state_groups_ids ) ) state = { } for st in state_groups_ids . itervalues ( ) : for key , e_id in st . iteritems ( ) : state . setdefault ( key , set ( ) ) . add ( e_id ) conflicted_state = { k : list ( v ) for k , v in state . iteritems ( ) if len ( v ) > 1 } if conflicted_state : logger . info ( "Resolving conflicted state for %r" , room_id ) with Measure ( self . clock , "state._resolve_events" ) : new_state = yield resolve_events_with_factory ( state_groups_ids . values ( ) , event_map = event_map , state_map_factory = state_map_factory , ) else : new_state = { key : e_ids . pop ( ) for key , e_ids in state . iteritems ( ) } with Measure ( self . clock , "state.create_group_ids" ) : state_group = None new_state_event_ids = frozenset ( new_state . itervalues ( ) ) for sg , events in state_groups_ids . iteritems ( ) : if new_state_event_ids == frozenset ( e_id for e_id in events ) : state_group = sg break prev_group = None delta_ids = None for old_group , old_ids in state_groups_ids . iteritems ( ) : if not set ( new_state ) - set ( old_ids ) : n_delta_ids = { k : v for k , v in new_state . iteritems ( ) if old_ids . get ( k ) != v } if not delta_ids or len ( n_delta_ids ) < len ( delta_ids ) : prev_group = old_group delta_ids = n_delta_ids cache = _StateCacheEntry ( state = new_state , state_group = state_group , prev_group = prev_group , delta_ids = delta_ids , ) if self . _state_cache is not None : self . _state_cache [ group_names ] = cache defer . returnValue ( cache )
Resolves conflicts between a set of state groups
Always generates a new state group (unless we hit the cache), so should
not be called for a single state group
Resolves conflicts between a set of state groups Always generates a new state group ( unless we hit the cache ) so should not be called for a single state group
"""Resolves conflicts between a set of state groups

        Always generates a new state group (unless we hit the cache), so should
        not be called for a single state group

        Args:
            room_id (str): room we are resolving for (used for logging)
            state_groups_ids (dict[int, dict[(str, str), str]]):
                 map from state group id to the state in that state group
                (where 'state' is a map from state key to event id)

            event_map(dict[str,FrozenEvent]|None):
                a dict from event_id to event, for any events that we happen to
                have in flight (eg, those currently being persisted). This will be
                used as a starting point fof finding the state we need; any missing
                events will be requested via state_map_factory.

                If None, all events will be fetched via state_map_factory.

        Returns:
            Deferred[_StateCacheEntry]: resolved state
        """ # build a map from state key to the event_ids which set that state. # dict[(str, str), set[str]) # build a map from state key to the event_ids which set that state, # including only those where there are state keys in conflict. # if the new state matches any of the input state groups, we can # use that state group again. Otherwise we will generate a state_id # which will be used as a cache key for future resolutions, but # not get persisted. # TODO: We want to create a state group for this set of events, to # increase cache hits, but we need to make sure that it doesn't # end up as a prev_group without being added to the database
None
Deferred[_StateCacheEntry]
self
room_id
room we are resolving for (used for logging)
room we are resolving for ( used for logging )
state_groups_ids
map from state group id to the state in that state group
(where 'state' is a map from state key to event id)
map from state group id to the state in that state group ( where ' state ' is a map from state key to event id )
event_map
a dict from event_id to event, for any events that we happen to
have in flight . This will be
used as a starting point fof finding the state we need; any missing
events will be requested via state_map_factory.

If None, all events will be fetched via state_map_factory.
a dict from event_id to event for any events that we happen to have in flight . This will be used as a starting point fof finding the state we need ; any missing events will be requested via state_map_factory . If None all events will be fetched via state_map_factory .
state_map_factory