synapse/storage/state.py
def _background_deduplicate_state(self, progress, batch_size):
        """This background update will slowly deduplicate state by reencoding
        them as deltas.
        """
        last_state_group = progress.get("last_state_group", 0)
        rows_inserted = progress.get("rows_inserted", 0)
        max_group = progress.get("max_group", None)

        BATCH_SIZE_SCALE_FACTOR = 100

        batch_size = max(1, int(batch_size / BATCH_SIZE_SCALE_FACTOR))

        if max_group is None:
            rows = yield self._execute(
                "_background_deduplicate_state", None,
                "SELECT coalesce(max(id), 0) FROM state_groups",
            )
            max_group = rows[0][0]

        def reindex_txn(txn):
            new_last_state_group = last_state_group
            for count in xrange(batch_size):
                txn.execute(
                    "SELECT id, room_id FROM state_groups"
                    " WHERE ? < id AND id <= ?"
                    " ORDER BY id ASC"
                    " LIMIT 1",
                    (new_last_state_group, max_group,)
                )
                row = txn.fetchone()
                if row:
                    state_group, room_id = row

                if not row or not state_group:
                    return True, count

                txn.execute(
                    "SELECT state_group FROM state_group_edges"
                    " WHERE state_group = ?",
                    (state_group,)
                )

                # If we reach a point where we've already started inserting
                # edges we should stop.
                if txn.fetchall():
                    return True, count

                txn.execute(
                    "SELECT coalesce(max(id), 0) FROM state_groups"
                    " WHERE id < ? AND room_id = ?",
                    (state_group, room_id,)
                )
                prev_group, = txn.fetchone()
                new_last_state_group = state_group

                if prev_group:
                    potential_hops = self._count_state_group_hops_txn(
                        txn, prev_group
                    )
                    if potential_hops >= MAX_STATE_DELTA_HOPS:
                        # We want to ensure chains are at most this long,#
                        # otherwise read performance degrades.
                        continue

                    prev_state = self._get_state_groups_from_groups_txn(
                        txn, [prev_group], types=None
                    )
                    prev_state = prev_state[prev_group]

                    curr_state = self._get_state_groups_from_groups_txn(
                        txn, [state_group], types=None
                    )
                    curr_state = curr_state[state_group]

                    if not set(prev_state.keys()) - set(curr_state.keys()):
                        # We can only do a delta if the current has a strict super set
                        # of keys

                        delta_state = {
                            key: value for key, value in curr_state.iteritems()
                            if prev_state.get(key, None) != value
                        }

                        self._simple_delete_txn(
                            txn,
                            table="state_group_edges",
                            keyvalues={
                                "state_group": state_group,
                            }
                        )

                        self._simple_insert_txn(
                            txn,
                            table="state_group_edges",
                            values={
                                "state_group": state_group,
                                "prev_state_group": prev_group,
                            }
                        )

                        self._simple_delete_txn(
                            txn,
                            table="state_groups_state",
                            keyvalues={
                                "state_group": state_group,
                            }
                        )

                        self._simple_insert_many_txn(
                            txn,
                            table="state_groups_state",
                            values=[
                                {
                                    "state_group": state_group,
                                    "room_id": room_id,
                                    "type": key[0],
                                    "state_key": key[1],
                                    "event_id": state_id,
                                }
                                for key, state_id in delta_state.iteritems()
                            ],
                        )

            progress = {
                "last_state_group": state_group,
                "rows_inserted": rows_inserted + batch_size,
                "max_group": max_group,
            }

            self._background_update_progress_txn(
                txn, self.STATE_GROUP_DEDUPLICATION_UPDATE_NAME, progress
            )

            return False, batch_size

        finished, result = yield self.runInteraction(
            self.STATE_GROUP_DEDUPLICATION_UPDATE_NAME, reindex_txn
        )

        if finished:
            yield self._end_background_update(self.STATE_GROUP_DEDUPLICATION_UPDATE_NAME)

        defer.returnValue(result * BATCH_SIZE_SCALE_FACTOR)
This background update will slowly deduplicate state by reencoding
        them as deltas.
        
This background update will slowly deduplicate state by reencoding
them as deltas.
This background update will slowly deduplicate state by reencoding them as deltas .
def _background_deduplicate_state(self, progress, batch_size):
        last_state_group = progress.get("last_state_group", 0)
        rows_inserted = progress.get("rows_inserted", 0)
        max_group = progress.get("max_group", None)
        BATCH_SIZE_SCALE_FACTOR = 100
        batch_size = max(1, int(batch_size / BATCH_SIZE_SCALE_FACTOR))
        if max_group is None:
            rows = yield self._execute(
                "_background_deduplicate_state", None,
                "SELECT coalesce(max(id), 0) FROM state_groups",
            )
            max_group = rows[0][0]
        def reindex_txn(txn):
            new_last_state_group = last_state_group
            for count in xrange(batch_size):
                txn.execute(
                    "SELECT id, room_id FROM state_groups"
                    " WHERE ? < id AND id <= ?"
                    " ORDER BY id ASC"
                    " LIMIT 1",
                    (new_last_state_group, max_group,)
                )
                row = txn.fetchone()
                if row:
                    state_group, room_id = row
                if not row or not state_group:
                    return True, count
                txn.execute(
                    "SELECT state_group FROM state_group_edges"
                    " WHERE state_group = ?",
                    (state_group,)
                )
                if txn.fetchall():
                    return True, count
                txn.execute(
                    "SELECT coalesce(max(id), 0) FROM state_groups"
                    " WHERE id < ? AND room_id = ?",
                    (state_group, room_id,)
                )
                prev_group, = txn.fetchone()
                new_last_state_group = state_group
                if prev_group:
                    potential_hops = self._count_state_group_hops_txn(
                        txn, prev_group
                    )
                    if potential_hops >= MAX_STATE_DELTA_HOPS:
                        continue
                    prev_state = self._get_state_groups_from_groups_txn(
                        txn, [prev_group], types=None
                    )
                    prev_state = prev_state[prev_group]
                    curr_state = self._get_state_groups_from_groups_txn(
                        txn, [state_group], types=None
                    )
                    curr_state = curr_state[state_group]
                    if not set(prev_state.keys()) - set(curr_state.keys()):
                        delta_state = {
                            key: value for key, value in curr_state.iteritems()
                            if prev_state.get(key, None) != value
                        }
                        self._simple_delete_txn(
                            txn,
                            table="state_group_edges",
                            keyvalues={
                                "state_group": state_group,
                            }
                        )
                        self._simple_insert_txn(
                            txn,
                            table="state_group_edges",
                            values={
                                "state_group": state_group,
                                "prev_state_group": prev_group,
                            }
                        )
                        self._simple_delete_txn(
                            txn,
                            table="state_groups_state",
                            keyvalues={
                                "state_group": state_group,
                            }
                        )
                        self._simple_insert_many_txn(
                            txn,
                            table="state_groups_state",
                            values=[
                                {
                                    "state_group": state_group,
                                    "room_id": room_id,
                                    "type": key[0],
                                    "state_key": key[1],
                                    "event_id": state_id,
                                }
                                for key, state_id in delta_state.iteritems()
                            ],
                        )
            progress = {
                "last_state_group": state_group,
                "rows_inserted": rows_inserted + batch_size,
                "max_group": max_group,
            }
            self._background_update_progress_txn(
                txn, self.STATE_GROUP_DEDUPLICATION_UPDATE_NAME, progress
            )
            return False, batch_size
        finished, result = yield self.runInteraction(
            self.STATE_GROUP_DEDUPLICATION_UPDATE_NAME, reindex_txn
        )
        if finished:
            yield self._end_background_update(self.STATE_GROUP_DEDUPLICATION_UPDATE_NAME)
        defer.returnValue(result * BATCH_SIZE_SCALE_FACTOR)
def _background_deduplicate_state ( self , progress , batch_size ) : last_state_group = progress . get ( "last_state_group" , 0 ) rows_inserted = progress . get ( "rows_inserted" , 0 ) max_group = progress . get ( "max_group" , None ) BATCH_SIZE_SCALE_FACTOR = 100 batch_size = max ( 1 , int ( batch_size / BATCH_SIZE_SCALE_FACTOR ) ) if max_group is None : rows = yield self . _execute ( "_background_deduplicate_state" , None , "SELECT coalesce(max(id), 0) FROM state_groups" , ) max_group = rows [ 0 ] [ 0 ] def reindex_txn ( txn ) : new_last_state_group = last_state_group for count in xrange ( batch_size ) : txn . execute ( "SELECT id, room_id FROM state_groups" " WHERE ? < id AND id <= ?" " ORDER BY id ASC" " LIMIT 1" , ( new_last_state_group , max_group , ) ) row = txn . fetchone ( ) if row : state_group , room_id = row if not row or not state_group : return True , count txn . execute ( "SELECT state_group FROM state_group_edges" " WHERE state_group = ?" , ( state_group , ) ) if txn . fetchall ( ) : return True , count txn . execute ( "SELECT coalesce(max(id), 0) FROM state_groups" " WHERE id < ? AND room_id = ?" , ( state_group , room_id , ) ) prev_group , = txn . fetchone ( ) new_last_state_group = state_group if prev_group : potential_hops = self . _count_state_group_hops_txn ( txn , prev_group ) if potential_hops >= MAX_STATE_DELTA_HOPS : continue prev_state = self . _get_state_groups_from_groups_txn ( txn , [ prev_group ] , types = None ) prev_state = prev_state [ prev_group ] curr_state = self . _get_state_groups_from_groups_txn ( txn , [ state_group ] , types = None ) curr_state = curr_state [ state_group ] if not set ( prev_state . keys ( ) ) - set ( curr_state . keys ( ) ) : delta_state = { key : value for key , value in curr_state . iteritems ( ) if prev_state . get ( key , None ) != value } self . _simple_delete_txn ( txn , table = "state_group_edges" , keyvalues = { "state_group" : state_group , } ) self . _simple_insert_txn ( txn , table = "state_group_edges" , values = { "state_group" : state_group , "prev_state_group" : prev_group , } ) self . _simple_delete_txn ( txn , table = "state_groups_state" , keyvalues = { "state_group" : state_group , } ) self . _simple_insert_many_txn ( txn , table = "state_groups_state" , values = [ { "state_group" : state_group , "room_id" : room_id , "type" : key [ 0 ] , "state_key" : key [ 1 ] , "event_id" : state_id , } for key , state_id in delta_state . iteritems ( ) ] , ) progress = { "last_state_group" : state_group , "rows_inserted" : rows_inserted + batch_size , "max_group" : max_group , } self . _background_update_progress_txn ( txn , self . STATE_GROUP_DEDUPLICATION_UPDATE_NAME , progress ) return False , batch_size finished , result = yield self . runInteraction ( self . STATE_GROUP_DEDUPLICATION_UPDATE_NAME , reindex_txn ) if finished : yield self . _end_background_update ( self . STATE_GROUP_DEDUPLICATION_UPDATE_NAME ) defer . returnValue ( result * BATCH_SIZE_SCALE_FACTOR )
This background update will slowly deduplicate state by reencoding
them as deltas.
This background update will slowly deduplicate state by reencoding them as deltas .
"""This background update will slowly deduplicate state by reencoding
        them as deltas.
        """ # If we reach a point where we've already started inserting # edges we should stop. # We want to ensure chains are at most this long,# # otherwise read performance degrades. # We can only do a delta if the current has a strict super set # of keys
self
progress
batch_size