watcha
/
watcha-synapse
mirror of https://github.com/watcha-fr/synapse
1
1
Fork 0
Code Issues Projects Releases Wiki Activity

Merge pull request #28 from matrix-org/erikj-perf

Browse Source 
Database performance improvements.
pull/4/merge
Mark Haines 10 years ago
parent adb04b1e57 af1c7c7808
commit 5e23a19204
5 changed files with 118 additions and 111 deletions
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Show Stats Download Patch File Download Diff File
  1. 13
      synapse/handlers/room.py
  2. 36
      synapse/storage/_base.py
  3. 21
      synapse/storage/roommember.py
  4. 14
      synapse/storage/state.py
  5. 145
      synapse/storage/stream.py

13
synapse/handlers/room.py
Unescape View File

@ -245,14 +245,12 @@ class RoomMemberHandler(BaseHandler):
self.distributor.declare("user_left_room")
@defer.inlineCallbacks
def get_room_members(self, room_id, membership=Membership.JOIN):
def get_room_members(self, room_id):
hs = self.hs
memberships = yield self.store.get_room_members(
room_id=room_id, membership=membership
)
users = yield self.store.get_users_in_room(room_id)
defer.returnValue([hs.parse_userid(m.user_id) for m in memberships])
defer.returnValue([hs.parse_userid(u) for u in users])
@defer.inlineCallbacks
def fetch_room_distributions_into(self, room_id, localusers=None,
@ -531,11 +529,10 @@ class RoomListHandler(BaseHandler):
def get_public_room_list(self):
chunk = yield self.store.get_rooms(is_public=True)
for room in chunk:
joined_members = yield self.store.get_room_members(
joined_users = yield self.store.get_users_in_room(
room_id=room["room_id"],
membership=Membership.JOIN
)
room["num_joined_members"] = len(joined_members)
room["num_joined_members"] = len(joined_users)
# FIXME (erikj): START is no longer a valid value
defer.returnValue({"start": "START", "end": "END", "chunk": chunk})

36
synapse/storage/_base.py
Unescape View File

@ -434,23 +434,29 @@ class SQLBaseStore(object):
return self.runInteraction("_simple_max_id", func)
def _get_events(self, event_ids):
def _get_events(self, event_ids, check_redacted=True,
get_prev_content=False):
return self.runInteraction(
"_get_events", self._get_events_txn, event_ids
"_get_events", self._get_events_txn, event_ids,
check_redacted=check_redacted, get_prev_content=get_prev_content,
)
def _get_events_txn(self, txn, event_ids):
events = []
for e_id in event_ids:
ev = self._get_event_txn(txn, e_id)
def _get_events_txn(self, txn, event_ids, check_redacted=True,
get_prev_content=False):
if not event_ids:
return []
if ev:
events.append(ev)
return events
return [
self._get_event_txn(
txn, event_id,
check_redacted=check_redacted,
get_prev_content=get_prev_content
)
for event_id in event_ids
]
def _get_event_txn(self, txn, event_id, check_redacted=True,
get_prev_content=True):
get_prev_content=False):
sql = (
"SELECT internal_metadata, json, r.event_id FROM event_json as e "
"LEFT JOIN redactions as r ON e.event_id = r.redacts "
@ -467,6 +473,14 @@ class SQLBaseStore(object):
internal_metadata, js, redacted = res
return self._get_event_from_row_txn(
txn, internal_metadata, js, redacted,
check_redacted=check_redacted,
get_prev_content=get_prev_content,
)
def _get_event_from_row_txn(self, txn, internal_metadata, js, redacted,
check_redacted=True, get_prev_content=False):
d = json.loads(js)
internal_metadata = json.loads(internal_metadata)

21
synapse/storage/roommember.py
Unescape View File

@ -123,6 +123,19 @@ class RoomMemberStore(SQLBaseStore):
else:
return None
def get_users_in_room(self, room_id):
def f(txn):
sql = (
"SELECT m.user_id FROM room_memberships as m"
" INNER JOIN current_state_events as c"
" ON m.event_id = c.event_id"
" WHERE m.membership = ? AND m.room_id = ?"
)
txn.execute(sql, (Membership.JOIN, room_id))
return [r[0] for r in txn.fetchall()]
return self.runInteraction("get_users_in_room", f)
def get_room_members(self, room_id, membership=None):
"""Retrieve the current room member list for a room.
@ -183,20 +196,14 @@ class RoomMemberStore(SQLBaseStore):
)
def _get_members_query_txn(self, txn, where_clause, where_values):
del_sql = (
"SELECT event_id FROM redactions WHERE redacts = e.event_id "
"LIMIT 1"
)
sql = (
"SELECT e.*, (%(redacted)s) AS redacted FROM events as e "
"SELECT e.* FROM events as e "
"INNER JOIN room_memberships as m "
"ON e.event_id = m.event_id "
"INNER JOIN current_state_events as c "
"ON m.event_id = c.event_id "
"WHERE %(where)s "
) % {
"redacted": del_sql,
"where": where_clause,
}

14
synapse/storage/state.py
Unescape View File

@ -15,6 +15,10 @@
from ._base import SQLBaseStore
import logging
logger = logging.getLogger(__name__)
class StateStore(SQLBaseStore):
""" Keeps track of the state at a given event.
@ -62,14 +66,8 @@ class StateStore(SQLBaseStore):
keyvalues={"state_group": group},
retcol="event_id",
)
state = []
for state_id in state_ids:
s = self._get_events_txn(
txn,
[state_id],
)
if s:
state.extend(s)
state = self._get_events_txn(txn, state_ids)
res[group] = state

145
synapse/storage/stream.py
Unescape View File

@ -137,7 +137,6 @@ class StreamStore(SQLBaseStore):
with_feedback=with_feedback,
)
@defer.inlineCallbacks
@log_function
def get_room_events_stream(self, user_id, from_key, to_key, room_id,
limit=0, with_feedback=False):
@ -157,11 +156,6 @@ class StreamStore(SQLBaseStore):
"WHERE m.user_id = ? "
)
del_sql = (
"SELECT event_id FROM redactions WHERE redacts = e.event_id "
"LIMIT 1"
)
if limit:
limit = max(limit, MAX_STREAM_SIZE)
else:
@ -172,38 +166,42 @@ class StreamStore(SQLBaseStore):
to_id = _parse_stream_token(to_key)
if from_key == to_key:
defer.returnValue(([], to_key))
return
return defer.succeed(([], to_key))
sql = (
"SELECT *, (%(redacted)s) AS redacted FROM events AS e WHERE "
"SELECT e.event_id, e.stream_ordering FROM events AS e WHERE "
"(e.outlier = 0 AND (room_id IN (%(current)s)) OR "
"(event_id IN (%(invites)s))) "
"AND e.stream_ordering > ? AND e.stream_ordering <= ? "
"ORDER BY stream_ordering ASC LIMIT %(limit)d "
) % {
"redacted": del_sql,
"current": current_room_membership_sql,
"invites": membership_sql,
"limit": limit
}
rows = yield self._execute_and_decode(
sql,
user_id, user_id, from_id, to_id
)
def f(txn):
txn.execute(sql, (user_id, user_id, from_id, to_id,))
ret = yield self._parse_events(rows)
rows = self.cursor_to_dict(txn)
if rows:
key = "s%d" % max([r["stream_ordering"] for r in rows])
else:
# Assume we didn't get anything because there was nothing to get.
key = to_key
ret = self._get_events_txn(
txn,
[r["event_id"] for r in rows],
get_prev_content=True
)
if rows:
key = "s%d" % max([r["stream_ordering"] for r in rows])
else:
# Assume we didn't get anything because there was nothing to
# get.
key = to_key
return ret, key
defer.returnValue((ret, key))
return self.runInteraction("get_room_events_stream", f)
@defer.inlineCallbacks
@log_function
def paginate_room_events(self, room_id, from_key, to_key=None,
direction='b', limit=-1,
@ -221,7 +219,9 @@ class StreamStore(SQLBaseStore):
bounds = _get_token_bound(from_key, from_comp)
if to_key:
bounds = "%s AND %s" % (bounds, _get_token_bound(to_key, to_comp))
bounds = "%s AND %s" % (
bounds, _get_token_bound(to_key, to_comp)
)
if int(limit) > 0:
args.append(int(limit))
@ -229,87 +229,78 @@ class StreamStore(SQLBaseStore):
else:
limit_str = ""
del_sql = (
"SELECT event_id FROM redactions WHERE redacts = events.event_id "
"LIMIT 1"
)
sql = (
"SELECT *, (%(redacted)s) AS redacted FROM events"
"SELECT * FROM events"
" WHERE outlier = 0 AND room_id = ? AND %(bounds)s"
" ORDER BY topological_ordering %(order)s,"
" stream_ordering %(order)s %(limit)s"
) % {
"redacted": del_sql,
"bounds": bounds,
"order": order,
"limit": limit_str
}
rows = yield self._execute_and_decode(
sql,
*args
)
if rows:
topo = rows[-1]["topological_ordering"]
toke = rows[-1]["stream_ordering"]
if direction == 'b':
topo -= 1
toke -= 1
next_token = "t%s-%s" % (topo, toke)
else:
# TODO (erikj): We should work out what to do here instead.
next_token = to_key if to_key else from_key
def f(txn):
txn.execute(sql, args)
rows = self.cursor_to_dict(txn)
if rows:
topo = rows[-1]["topological_ordering"]
toke = rows[-1]["stream_ordering"]
if direction == 'b':
topo -= 1
toke -= 1
next_token = "t%s-%s" % (topo, toke)
else:
# TODO (erikj): We should work out what to do here instead.
next_token = to_key if to_key else from_key
events = self._get_events_txn(
txn,
[r["event_id"] for r in rows],
get_prev_content=True
)
events = yield self._parse_events(rows)
return events, next_token,
defer.returnValue(
(
events,
next_token
)
)
return self.runInteraction("paginate_room_events", f)
@defer.inlineCallbacks
def get_recent_events_for_room(self, room_id, limit, end_token,
with_feedback=False):
# TODO (erikj): Handle compressed feedback
del_sql = (
"SELECT event_id FROM redactions WHERE redacts = events.event_id "
"LIMIT 1"
)
sql = (
"SELECT *, (%(redacted)s) AS redacted FROM events "
"SELECT * FROM events "
"WHERE room_id = ? AND stream_ordering <= ? AND outlier = 0 "
"ORDER BY topological_ordering DESC, stream_ordering DESC LIMIT ? "
) % {
"redacted": del_sql,
}
rows = yield self._execute_and_decode(
sql,
room_id, end_token, limit
)
rows.reverse() # As we selected with reverse ordering
def f(txn):
txn.execute(sql, (room_id, end_token, limit,))
if rows:
topo = rows[0]["topological_ordering"]
toke = rows[0]["stream_ordering"]
start_token = "t%s-%s" % (topo, toke)
rows = self.cursor_to_dict(txn)
token = (start_token, end_token)
else:
token = (end_token, end_token)
rows.reverse() # As we selected with reverse ordering
events = yield self._parse_events(rows)
if rows:
topo = rows[0]["topological_ordering"]
toke = rows[0]["stream_ordering"]
start_token = "t%s-%s" % (topo, toke)
token = (start_token, end_token)
else:
token = (end_token, end_token)
events = self._get_events_txn(
txn,
[r["event_id"] for r in rows],
get_prev_content=True
)
ret = (events, token)
return events, token
defer.returnValue(ret)
return self.runInteraction("get_recent_events_for_room", f)
def get_room_events_max_id(self):
return self.runInteraction(

Write Preview
Loading…
Cancel
Save