Add stream change cache

This commit is contained in:
Erik Johnston 2017-05-31 15:46:36 +01:00
parent 5d79d728f5
commit 304880d185
4 changed files with 34 additions and 1 deletions

View File

@ -223,6 +223,18 @@ class DataStore(RoomMemberStore, RoomStore,
"DeviceListFederationStreamChangeCache", device_list_max,
)
curr_state_delta_prefill, min_curr_state_delta_id = self._get_cache_dict(
db_conn, "current_state_delta_stream",
entity_column="room_id",
stream_column="stream_id",
max_value=events_max, # As we share the stream id with events token
limit=1000,
)
self._curr_state_delta_stream_cache = StreamChangeCache(
"_curr_state_delta_stream_cache", min_curr_state_delta_id,
prefilled_cache=curr_state_delta_prefill,
)
cur = LoggingTransaction(
db_conn.cursor(),
name="_find_stream_orderings_for_times_txn",

View File

@ -755,6 +755,10 @@ class EventsStore(SQLBaseStore):
]
)
self._curr_state_delta_stream_cache.enttity_has_changed(
room_id, max_stream_order,
)
# Invalidate the various caches
# Figure out the changes of membership to invalidate the

View File

@ -204,7 +204,9 @@ class UserDirectoryStore(SQLBaseStore):
)
def get_current_state_deltas(self, prev_stream_id):
# TODO: Add stream change cache
if not self._curr_state_delta_stream_cache.has_any_entity_changed(prev_stream_id):
return []
# TODO: Add limit
sql = """
SELECT stream_id, room_id, type, state_key, event_id, prev_event_id

View File

@ -89,6 +89,21 @@ class StreamChangeCache(object):
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()
if stream_pos >= max(self._cache):
return False
else:
return True
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.