summary refs log tree commit diff
path: root/synapse
diff options
context:
space:
mode:
Diffstat (limited to 'synapse')
-rw-r--r--synapse/handlers/federation.py8
-rw-r--r--synapse/handlers/message.py6
-rw-r--r--synapse/handlers/sync.py6
-rw-r--r--synapse/state.py10
-rw-r--r--synapse/storage/_base.py43
-rw-r--r--synapse/storage/state.py267
-rw-r--r--synapse/storage/stream.py3
-rw-r--r--synapse/util/dictionary_cache.py106
8 files changed, 351 insertions, 98 deletions
diff --git a/synapse/handlers/federation.py b/synapse/handlers/federation.py
index f7155fd8d3..90649af9e1 100644
--- a/synapse/handlers/federation.py
+++ b/synapse/handlers/federation.py
@@ -230,7 +230,11 @@ class FederationHandler(BaseHandler):
     @defer.inlineCallbacks
     def _filter_events_for_server(self, server_name, room_id, events):
         states = yield self.store.get_state_for_events(
-            room_id, [e.event_id for e in events],
+            room_id, frozenset(e.event_id for e in events),
+            types=(
+                (EventTypes.RoomHistoryVisibility, ""),
+                (EventTypes.Member, None),
+            )
         )
 
         events_and_states = zip(events, states)
@@ -503,7 +507,7 @@ class FederationHandler(BaseHandler):
         event_ids = list(extremities.keys())
 
         states = yield defer.gatherResults([
-            self.state_handler.resolve_state_groups([e])
+            self.state_handler.resolve_state_groups(room_id, [e])
             for e in event_ids
         ])
         states = dict(zip(event_ids, [s[1] for s in states]))
diff --git a/synapse/handlers/message.py b/synapse/handlers/message.py
index 9d6d4f0978..765b14d994 100644
--- a/synapse/handlers/message.py
+++ b/synapse/handlers/message.py
@@ -138,7 +138,11 @@ class MessageHandler(BaseHandler):
     @defer.inlineCallbacks
     def _filter_events_for_client(self, user_id, room_id, events):
         states = yield self.store.get_state_for_events(
-            room_id, [e.event_id for e in events],
+            room_id, frozenset(e.event_id for e in events),
+            types=(
+                (EventTypes.RoomHistoryVisibility, ""),
+                (EventTypes.Member, user_id),
+            )
         )
 
         events_and_states = zip(events, states)
diff --git a/synapse/handlers/sync.py b/synapse/handlers/sync.py
index 6cff6230c1..8f58774b31 100644
--- a/synapse/handlers/sync.py
+++ b/synapse/handlers/sync.py
@@ -295,7 +295,11 @@ class SyncHandler(BaseHandler):
     @defer.inlineCallbacks
     def _filter_events_for_client(self, user_id, room_id, events):
         states = yield self.store.get_state_for_events(
-            room_id, [e.event_id for e in events],
+            room_id, frozenset(e.event_id for e in events),
+            types=(
+                (EventTypes.RoomHistoryVisibility, ""),
+                (EventTypes.Member, user_id),
+            )
         )
 
         events_and_states = zip(events, states)
diff --git a/synapse/state.py b/synapse/state.py
index 80da90a72c..b5e5d7bbda 100644
--- a/synapse/state.py
+++ b/synapse/state.py
@@ -96,7 +96,7 @@ class StateHandler(object):
             cache.ts = self.clock.time_msec()
             state = cache.state
         else:
-            res = yield self.resolve_state_groups(event_ids)
+            res = yield self.resolve_state_groups(room_id, event_ids)
             state = res[1]
 
         if event_type:
@@ -155,13 +155,13 @@ class StateHandler(object):
 
         if event.is_state():
             ret = yield self.resolve_state_groups(
-                [e for e, _ in event.prev_events],
+                event.room_id, [e for e, _ in event.prev_events],
                 event_type=event.type,
                 state_key=event.state_key,
             )
         else:
             ret = yield self.resolve_state_groups(
-                [e for e, _ in event.prev_events],
+                event.room_id, [e for e, _ in event.prev_events],
             )
 
         group, curr_state, prev_state = ret
@@ -180,7 +180,7 @@ class StateHandler(object):
 
     @defer.inlineCallbacks
     @log_function
-    def resolve_state_groups(self, event_ids, event_type=None, state_key=""):
+    def resolve_state_groups(self, room_id, event_ids, event_type=None, state_key=""):
         """ Given a list of event_ids this method fetches the state at each
         event, resolves conflicts between them and returns them.
 
@@ -205,7 +205,7 @@ class StateHandler(object):
                 )
 
         state_groups = yield self.store.get_state_groups(
-            event_ids
+            room_id, event_ids
         )
 
         logger.debug(
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index 2a601e37e3..30a9e3f0a3 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -18,6 +18,7 @@ from synapse.api.errors import StoreError
 from synapse.util.logutils import log_function
 from synapse.util.logcontext import preserve_context_over_fn, LoggingContext
 from synapse.util.lrucache import LruCache
+from synapse.util.dictionary_cache import DictionaryCache
 import synapse.metrics
 
 from util.id_generators import IdGenerator, StreamIdGenerator
@@ -71,6 +72,11 @@ class Cache(object):
         self.thread = None
         caches_by_name[name] = self.cache
 
+        class Sentinel(object):
+            __slots__ = []
+
+        self.sentinel = Sentinel()
+
     def check_thread(self):
         expected_thread = self.thread
         if expected_thread is None:
@@ -82,22 +88,33 @@ class Cache(object):
                 )
 
     def get(self, *keyargs):
-        if len(keyargs) != self.keylen:
-            raise ValueError("Expected a key to have %d items", self.keylen)
+        try:
+            if len(keyargs) != self.keylen:
+                raise ValueError("Expected a key to have %d items", self.keylen)
 
-        if keyargs in self.cache:
-            cache_counter.inc_hits(self.name)
-            return self.cache[keyargs]
+            val = self.cache.get(keyargs, self.sentinel)
+            if val is not self.sentinel:
+                cache_counter.inc_hits(self.name)
+                return val
 
-        cache_counter.inc_misses(self.name)
-        raise KeyError()
+            cache_counter.inc_misses(self.name)
+            raise KeyError()
+        except KeyError:
+            raise
+        except:
+            logger.exception("Cache.get failed for %s" % (self.name,))
+            raise
 
     def update(self, sequence, *args):
-        self.check_thread()
-        if self.sequence == sequence:
-            # Only update the cache if the caches sequence number matches the
-            # number that the cache had before the SELECT was started (SYN-369)
-            self.prefill(*args)
+        try:
+            self.check_thread()
+            if self.sequence == sequence:
+                # Only update the cache if the caches sequence number matches the
+                # number that the cache had before the SELECT was started (SYN-369)
+                self.prefill(*args)
+        except:
+            logger.exception("Cache.update failed for %s" % (self.name,))
+            raise
 
     def prefill(self, *args):  # because I can't  *keyargs, value
         keyargs = args[:-1]
@@ -321,6 +338,8 @@ class SQLBaseStore(object):
         self._get_event_cache = Cache("*getEvent*", keylen=3, lru=True,
                                       max_entries=hs.config.event_cache_size)
 
+        self._state_group_cache = DictionaryCache("*stateGroupCache*", 100000)
+
         self._event_fetch_lock = threading.Condition()
         self._event_fetch_list = []
         self._event_fetch_ongoing = 0
diff --git a/synapse/storage/state.py b/synapse/storage/state.py
index 47bec65497..a967b3d44b 100644
--- a/synapse/storage/state.py
+++ b/synapse/storage/state.py
@@ -17,6 +17,7 @@ from ._base import SQLBaseStore, cached
 
 from twisted.internet import defer
 
+from synapse.util import unwrapFirstError
 from synapse.util.stringutils import random_string
 
 import logging
@@ -44,52 +45,38 @@ class StateStore(SQLBaseStore):
     """
 
     @defer.inlineCallbacks
-    def get_state_groups(self, event_ids):
+    def get_state_groups(self, room_id, event_ids):
         """ Get the state groups for the given list of event_ids
 
         The return value is a dict mapping group names to lists of events.
         """
 
-        def f(txn):
-            groups = set()
-            for event_id in event_ids:
-                group = self._simple_select_one_onecol_txn(
-                    txn,
-                    table="event_to_state_groups",
-                    keyvalues={"event_id": event_id},
-                    retcol="state_group",
-                    allow_none=True,
-                )
-                if group:
-                    groups.add(group)
-
-            res = {}
-            for group in groups:
-                state_ids = self._simple_select_onecol_txn(
-                    txn,
-                    table="state_groups_state",
-                    keyvalues={"state_group": group},
-                    retcol="event_id",
-                )
-
-                res[group] = state_ids
-
-            return res
+        event_and_groups = yield defer.gatherResults(
+            [
+                self._get_state_group_for_event(
+                    room_id, event_id,
+                ).addCallback(lambda group, event_id: (event_id, group), event_id)
+                for event_id in event_ids
+            ],
+            consumeErrors=True,
+        ).addErrback(unwrapFirstError)
 
-        states = yield self.runInteraction(
-            "get_state_groups",
-            f,
-        )
+        groups = set(group for _, group in event_and_groups if group)
 
-        state_list = yield defer.gatherResults(
+        group_to_state = yield defer.gatherResults(
             [
-                self._fetch_events_for_group(group, vals)
-                for group, vals in states.items()
+                self._get_state_for_group(
+                    group,
+                ).addCallback(lambda state_dict, group: (group, state_dict), group)
+                for group in groups
             ],
             consumeErrors=True,
-        )
+        ).addErrback(unwrapFirstError)
 
-        defer.returnValue(dict(state_list))
+        defer.returnValue({
+            group: state_map.values()
+            for group, state_map in group_to_state
+        })
 
     @cached(num_args=1)
     def _fetch_events_for_group(self, key, events):
@@ -206,65 +193,195 @@ class StateStore(SQLBaseStore):
         events = yield self._get_events(event_ids, get_prev_content=False)
         defer.returnValue(events)
 
-    @defer.inlineCallbacks
-    def get_state_for_events(self, room_id, event_ids):
+    @cached(num_args=2, lru=True, max_entries=10000)
+    def _get_state_groups_from_group(self, group, types):
         def f(txn):
-            groups = set()
-            event_to_group = {}
-            for event_id in event_ids:
-                # TODO: Remove this loop.
-                group = self._simple_select_one_onecol_txn(
-                    txn,
-                    table="event_to_state_groups",
-                    keyvalues={"event_id": event_id},
-                    retcol="state_group",
-                    allow_none=True,
-                )
-                if group:
-                    event_to_group[event_id] = group
-                    groups.add(group)
-
-            group_to_state_ids = {}
-            for group in groups:
-                state_ids = self._simple_select_onecol_txn(
-                    txn,
-                    table="state_groups_state",
-                    keyvalues={"state_group": group},
-                    retcol="event_id",
+            if types is not None:
+                where_clause = "AND (%s)" % (
+                    " OR ".join(["(type = ? AND state_key = ?)"] * len(types)),
                 )
+            else:
+                where_clause = ""
+
+            sql = (
+                "SELECT event_id FROM state_groups_state WHERE"
+                " state_group = ? %s"
+            ) % (where_clause,)
+
+            args = [group]
+            if types is not None:
+                args.extend([i for typ in types for i in typ])
+
+            txn.execute(sql, args)
+
+            return group, [
+                r[0]
+                for r in txn.fetchall()
+            ]
+
+        return self.runInteraction(
+            "_get_state_groups_from_group",
+            f,
+        )
+
+    @cached(num_args=3, lru=True, max_entries=20000)
+    def _get_state_for_event_id(self, room_id, event_id, types):
+        def f(txn):
+            type_and_state_sql = " OR ".join([
+                "(type = ? AND state_key = ?)"
+                if typ[1] is not None
+                else "type = ?"
+                for typ in types
+            ])
 
-                group_to_state_ids[group] = state_ids
+            sql = (
+                "SELECT sg.event_id FROM state_groups_state as sg"
+                " INNER JOIN event_to_state_groups as e"
+                " ON e.state_group = sg.state_group"
+                " WHERE e.event_id = ? AND (%s)"
+            ) % (type_and_state_sql,)
+
+            args = [event_id]
+            for typ, state_key in types:
+                args.extend(
+                    [typ, state_key] if state_key is not None else [typ]
+                )
+            txn.execute(sql, args)
 
-            return event_to_group, group_to_state_ids
+            return event_id, [
+                r[0]
+                for r in txn.fetchall()
+            ]
 
-        res = yield self.runInteraction(
-            "annotate_events_with_state_groups",
+        return self.runInteraction(
+            "_get_state_for_event_id",
             f,
         )
 
-        event_to_group, group_to_state_ids = res
+    @defer.inlineCallbacks
+    def get_state_for_events(self, room_id, event_ids, types):
+        """Given a list of event_ids and type tuples, return a list of state
+        dicts for each event. The state dicts will only have the type/state_keys
+        that are in the `types` list.
+
+        Args:
+            room_id (str)
+            event_ids (list)
+            types (list): List of (type, state_key) tuples which are used to
+                filter the state fetched. `state_key` may be None, which matches
+                any `state_key`
+
+        Returns:
+            deferred: A list of dicts corresponding to the event_ids given.
+            The dicts are mappings from (type, state_key) -> state_events
+        """
+        event_and_groups = yield defer.gatherResults(
+            [
+                self._get_state_group_for_event(
+                    room_id, event_id,
+                ).addCallback(lambda group, event_id: (event_id, group), event_id)
+                for event_id in event_ids
+            ],
+            consumeErrors=True,
+        ).addErrback(unwrapFirstError)
+
+        groups = set(group for _, group in event_and_groups)
 
-        state_list = yield defer.gatherResults(
+        res = yield defer.gatherResults(
             [
-                self._fetch_events_for_group(group, vals)
-                for group, vals in group_to_state_ids.items()
+                self._get_state_for_group(
+                    group, types
+                ).addCallback(lambda state_dict, group: (group, state_dict), group)
+                for group in groups
             ],
             consumeErrors=True,
-        )
+        ).addErrback(unwrapFirstError)
 
-        state_dict = {
-            group: {
-                (ev.type, ev.state_key): ev
-                for ev in state
-            }
-            for group, state in state_list
+        group_to_state = dict(res)
+
+        event_to_state = {
+            event_id: group_to_state[group]
+            for event_id, group in event_and_groups
         }
 
         defer.returnValue([
-            state_dict.get(event_to_group.get(event, None), None)
+            event_to_state[event]
             for event in event_ids
         ])
 
+    @cached(num_args=2, lru=True, max_entries=100000)
+    def _get_state_group_for_event(self, room_id, event_id):
+        return self._simple_select_one_onecol(
+            table="event_to_state_groups",
+            keyvalues={
+                "event_id": event_id,
+            },
+            retcol="state_group",
+            allow_none=True,
+            desc="_get_state_group_for_event",
+        )
+
+    @defer.inlineCallbacks
+    def _get_state_for_group(self, group, types=None):
+        is_all, state_dict = self._state_group_cache.get(group)
+
+        type_to_key = {}
+        missing_types = set()
+        if types is not None:
+            for typ, state_key in types:
+                if state_key is None:
+                    type_to_key[typ] = None
+                    missing_types.add((typ, state_key))
+                else:
+                    if type_to_key.get(typ, object()) is not None:
+                        type_to_key.setdefault(typ, set()).add(state_key)
+
+                    if (typ, state_key) not in state_dict:
+                        missing_types.add((typ, state_key))
+
+        if is_all and types is None:
+            defer.returnValue(state_dict)
+
+        if is_all or (types is not None and not missing_types):
+            def include(typ, state_key):
+                sentinel = object()
+                valid_state_keys = type_to_key.get(typ, sentinel)
+                if valid_state_keys is sentinel:
+                    return False
+                if valid_state_keys is None:
+                    return True
+                if state_key in valid_state_keys:
+                    return True
+                return False
+
+            defer.returnValue({
+                k: v
+                for k, v in state_dict.items()
+                if include(k[0], k[1])
+            })
+
+        # Okay, so we have some missing_types, lets fetch them.
+        cache_seq_num = self._state_group_cache.sequence
+        _, state_ids = yield self._get_state_groups_from_group(
+            group,
+            frozenset(types) if types else None
+        )
+        state_events = yield self._get_events(state_ids, get_prev_content=False)
+        state_dict = {
+            (e.type, e.state_key): e
+            for e in state_events
+        }
+
+        # Update the cache
+        self._state_group_cache.update(
+            cache_seq_num,
+            key=group,
+            value=state_dict,
+            full=(types is None),
+        )
+
+        defer.returnValue(state_dict)
+
 
 def _make_group_id(clock):
     return str(int(clock.time_msec())) + random_string(5)
diff --git a/synapse/storage/stream.py b/synapse/storage/stream.py
index af45fc5619..9db259d5fc 100644
--- a/synapse/storage/stream.py
+++ b/synapse/storage/stream.py
@@ -300,8 +300,7 @@ class StreamStore(SQLBaseStore):
         defer.returnValue((events, token))
 
     @defer.inlineCallbacks
-    def get_recent_events_for_room(self, room_id, limit, end_token,
-                                   with_feedback=False, from_token=None):
+    def get_recent_events_for_room(self, room_id, limit, end_token, from_token=None):
         # TODO (erikj): Handle compressed feedback
 
         end_token = RoomStreamToken.parse_stream_token(end_token)
diff --git a/synapse/util/dictionary_cache.py b/synapse/util/dictionary_cache.py
new file mode 100644
index 0000000000..38b131677c
--- /dev/null
+++ b/synapse/util/dictionary_cache.py
@@ -0,0 +1,106 @@
+# -*- coding: utf-8 -*-
+# Copyright 2015 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.lrucache import LruCache
+from collections import namedtuple
+import threading
+import logging
+
+
+logger = logging.getLogger(__name__)
+
+
+DictionaryEntry = namedtuple("DictionaryEntry", ("full", "value"))
+
+
+class DictionaryCache(object):
+
+    def __init__(self, name, max_entries=1000):
+        self.cache = LruCache(max_size=max_entries)
+
+        self.name = name
+        self.sequence = 0
+        self.thread = None
+        # caches_by_name[name] = self.cache
+
+        class Sentinel(object):
+            __slots__ = []
+
+        self.sentinel = Sentinel()
+
+    def check_thread(self):
+        expected_thread = self.thread
+        if expected_thread is None:
+            self.thread = threading.current_thread()
+        else:
+            if expected_thread is not threading.current_thread():
+                raise ValueError(
+                    "Cache objects can only be accessed from the main thread"
+                )
+
+    def get(self, key, dict_keys=None):
+        try:
+            entry = self.cache.get(key, self.sentinel)
+            if entry is not self.sentinel:
+                # cache_counter.inc_hits(self.name)
+
+                if dict_keys is None:
+                    return DictionaryEntry(entry.full, dict(entry.value))
+                else:
+                    return DictionaryEntry(entry.full, {
+                        k: entry.value[k]
+                        for k in dict_keys
+                        if k in entry.value
+                    })
+
+            # cache_counter.inc_misses(self.name)
+            return DictionaryEntry(False, {})
+        except:
+            logger.exception("get failed")
+            raise
+
+    def invalidate(self, key):
+        self.check_thread()
+
+        # Increment the sequence number so that any SELECT statements that
+        # raced with the INSERT don't update the cache (SYN-369)
+        self.sequence += 1
+        self.cache.pop(key, None)
+
+    def invalidate_all(self):
+        self.check_thread()
+        self.sequence += 1
+        self.cache.clear()
+
+    def update(self, sequence, key, value, full=False):
+        try:
+            self.check_thread()
+            if self.sequence == sequence:
+                # Only update the cache if the caches sequence number matches the
+                # number that the cache had before the SELECT was started (SYN-369)
+                if full:
+                    self._insert(key, value)
+                else:
+                    self._update_or_insert(key, value)
+        except:
+            logger.exception("update failed")
+            raise
+
+    def _update_or_insert(self, key, value):
+        entry = self.cache.setdefault(key, DictionaryEntry(False, {}))
+        entry.value.update(value)
+
+    def _insert(self, key, value):
+        self.cache[key] = DictionaryEntry(True, value)