synapse/synapse/util/caches/stream_change_cache.py

# -*- coding: utf-8 -*-
# Copyright 2016 OpenMarket Ltd
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from synapse.util.caches import register_cache, CACHE_SIZE_FACTOR


from blist import sorteddict
import logging


logger = logging.getLogger(__name__)


class StreamChangeCache(object):
    """Keeps track of the stream positions of the latest change in a set of entities.

    Typically the entity will be a room or user id.

    Given a list of entities and a stream position, it will give a subset of
    entities that may have changed since that position. If position key is too
    old then the cache will simply return all given entities.
    """
    def __init__(self, name, current_stream_pos, max_size=10000, prefilled_cache={}):
        self._max_size = int(max_size * CACHE_SIZE_FACTOR)
        self._entity_to_key = {}
        self._cache = sorteddict()
        self._earliest_known_stream_pos = current_stream_pos
        self.name = name
        self.metrics = register_cache("cache", self.name, self._cache)

        for entity, stream_pos in prefilled_cache.items():
            self.entity_has_changed(entity, stream_pos)

    def has_entity_changed(self, entity, stream_pos):
        """Returns True if the entity may have been updated since stream_pos
        """
        assert type(stream_pos) is int or type(stream_pos) is long

        if stream_pos < self._earliest_known_stream_pos:
            self.metrics.inc_misses()
            return True

        latest_entity_change_pos = self._entity_to_key.get(entity, None)
        if latest_entity_change_pos is None:
            self.metrics.inc_hits()
            return False

        if stream_pos < latest_entity_change_pos:
            self.metrics.inc_misses()
            return True

        self.metrics.inc_hits()
        return False

    def get_entities_changed(self, entities, stream_pos):
        """Returns subset of entities that have had new things since the
        given position. If the position is too old it will just return the given list.
        """
        assert type(stream_pos) is int

        if stream_pos >= self._earliest_known_stream_pos:
            keys = self._cache.keys()
            i = keys.bisect_right(stream_pos)

            result = set(
                self._cache[k] for k in keys[i:]
            ).intersection(entities)

            self.metrics.inc_hits()
        else:
            result = entities
            self.metrics.inc_misses()

        return result

    def has_any_entity_changed(self, stream_pos):
        """Returns if any entity has changed
        """
        assert type(stream_pos) is int

        if stream_pos >= self._earliest_known_stream_pos:
            self.metrics.inc_hits()
            keys = self._cache.keys()
            i = keys.bisect_right(stream_pos)

            return i < len(keys)
        else:
            self.metrics.inc_misses()
            return True

    def get_all_entities_changed(self, stream_pos):
        """Returns all entites that have had new things since the given
        position. If the position is too old it will return None.
        """
        assert type(stream_pos) is int

        if stream_pos >= self._earliest_known_stream_pos:
            keys = self._cache.keys()
            i = keys.bisect_right(stream_pos)

            return [self._cache[k] for k in keys[i:]]
        else:
            return None

    def entity_has_changed(self, entity, stream_pos):
        """Informs the cache that the entity has been changed at the given
        position.
        """
        assert type(stream_pos) is int

        if stream_pos > self._earliest_known_stream_pos:
            old_pos = self._entity_to_key.get(entity, None)
            if old_pos is not None:
                stream_pos = max(stream_pos, old_pos)
                self._cache.pop(old_pos, None)
            self._cache[stream_pos] = entity
            self._entity_to_key[entity] = stream_pos

            while len(self._cache) > self._max_size:
                k, r = self._cache.popitem()
                self._earliest_known_stream_pos = max(k, self._earliest_known_stream_pos)
                self._entity_to_key.pop(r, None)

    def get_max_pos_of_last_change(self, entity):
        """Returns an upper bound of the stream id of the last change to an
        entity.
        """
        return self._entity_to_key.get(entity, self._earliest_known_stream_pos)