tileradotorg/synapse
1
0
Fork 0
forked from MirrorHub/synapse
Code Pull requests Activity

Merge pull request #3497 from matrix-org/rav/measure_fetch_event_loop

Browse source 
Add CPU metrics for _fetch_event_list
This commit is contained in:
Richard van der Hoff 2018-07-10 10:00:24 +01:00 committed by GitHub
commit b1fe697b3c
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
2 changed files with 33 additions and 19 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
changelog.d/3497.feature Normal file
View file

@ -0,0 +1 @@
Add CPU metrics for _fetch_event_list

19
synapse/storage/events_worker.py
View file

@ -222,10 +222,8 @@ class EventsWorkerStore(SQLBaseStore):
"""Takes a database connection and waits for requests for events from """Takes a database connection and waits for requests for events from
the _event_fetch_list queue. the _event_fetch_list queue.
""" """
event_list = []
i = 0 i = 0
while True: while True:
try:
with self._event_fetch_lock: with self._event_fetch_lock:
event_list = self._event_fetch_list event_list = self._event_fetch_list
self._event_fetch_list = [] self._event_fetch_list = []
@ -241,6 +239,22 @@ class EventsWorkerStore(SQLBaseStore):
continue continue
i = 0 i = 0
self._fetch_event_list(conn, event_list)
def _fetch_event_list(self, conn, event_list):
"""Handle a load of requests from the _event_fetch_list queue
Args:
conn (twisted.enterprise.adbapi.Connection): database connection
event_list (list[Tuple[list[str], Deferred]]):
The fetch requests. Each entry consists of a list of event
ids to be fetched, and a deferred to be completed once the
events have been fetched.
"""
with Measure(self._clock, "_fetch_event_list"):
try:
event_id_lists = zip(*event_list)[0] event_id_lists = zip(*event_list)[0]
event_ids = [ event_ids = [
item for sublist in event_id_lists for item in sublist item for sublist in event_id_lists for item in sublist
@ -280,7 +294,6 @@ class EventsWorkerStore(SQLBaseStore):
with PreserveLoggingContext(): with PreserveLoggingContext():
d.errback(e) d.errback(e)
if event_list:
with PreserveLoggingContext(): with PreserveLoggingContext():
self.hs.get_reactor().callFromThread(fire, event_list) self.hs.get_reactor().callFromThread(fire, event_list)