Skip to content

Commit

Permalink
Batch fetch _get_state_groups_from_groups
Browse files Browse the repository at this point in the history
  • Loading branch information
erikjohnston committed Feb 10, 2016
1 parent 24f00a6 commit 5189bfd
Showing 1 changed file with 34 additions and 32 deletions.
66 changes: 34 additions & 32 deletions synapse/storage/state.py
Original file line number Diff line number Diff line change
Expand Up @@ -171,41 +171,43 @@ def f(txn):
events = yield self._get_events(event_ids, get_prev_content=False)
defer.returnValue(events)

def _get_state_groups_from_groups(self, groups_and_types):
def _get_state_groups_from_groups(self, groups, types):
"""Returns dictionary state_group -> state event ids
Args:
groups_and_types (list): list of 2-tuple (`group`, `types`)
"""
def f(txn):
results = {}
for group, types in groups_and_types:
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,)
def f(txn, groups):
if types is not None:
where_clause = "AND (%s)" % (
" OR ".join(["(type = ? AND state_key = ?)"] * len(types)),
)
else:
where_clause = ""

args = [group]
if types is not None:
args.extend([i for typ in types for i in typ])
sql = (
"SELECT state_group, event_id FROM state_groups_state WHERE"
" state_group IN (%s) %s" % (
",".join("?" for _ in groups),
where_clause,
)
)

txn.execute(sql, args)
args = list(groups)
if types is not None:
args.extend([i for typ in types for i in typ])

results[group] = [r[0] for r in txn.fetchall()]
txn.execute(sql, args)
rows = self.cursor_to_dict(txn)

results = {}
for row in rows:
results.setdefault(row["state_group"], []).append(row["event_id"])
return results

return self.runInteraction(
"_get_state_groups_from_groups",
f,
)
chunks = [groups[i:i + 100] for i in xrange(0, len(groups), 100)]
for chunk in chunks:
return self.runInteraction(
"_get_state_groups_from_groups",
f, chunk
)

@defer.inlineCallbacks
def get_state_for_events(self, event_ids, types):
Expand Down Expand Up @@ -349,7 +351,7 @@ def _get_state_for_groups(self, groups, types=None):
all events are returned.
"""
results = {}
missing_groups_and_types = []
missing_groups = []
if types is not None:
for group in set(groups):
state_dict, missing_types, got_all = self._get_some_state_from_cache(
Expand All @@ -358,7 +360,7 @@ def _get_state_for_groups(self, groups, types=None):
results[group] = state_dict

if not got_all:
missing_groups_and_types.append((group, missing_types))
missing_groups.append(group)
else:
for group in set(groups):
state_dict, got_all = self._get_all_state_from_cache(
Expand All @@ -367,9 +369,9 @@ def _get_state_for_groups(self, groups, types=None):
results[group] = state_dict

if not got_all:
missing_groups_and_types.append((group, None))
missing_groups.append(group)

if not missing_groups_and_types:
if not missing_groups:
defer.returnValue({
group: {
type_tuple: event
Expand All @@ -383,7 +385,7 @@ def _get_state_for_groups(self, groups, types=None):
cache_seq_num = self._state_group_cache.sequence

group_state_dict = yield self._get_state_groups_from_groups(
missing_groups_and_types
missing_groups, types
)

state_events = yield self._get_events(
Expand Down

0 comments on commit 5189bfd

Please sign in to comment.