Count and loop
2 files changed, 35 insertions, 37 deletions
diff --git a/synapse/storage/_base.py b/synapse/storage/_base.py
index c20ff3a572..97bf42469a 100644
--- a/synapse/storage/_base.py
+++ b/synapse/storage/_base.py
@@ -301,7 +301,7 @@ class SQLBaseStore(object):
self._event_fetch_lock = threading.Lock()
self._event_fetch_list = []
- self._event_fetch_ongoing = False
+ self._event_fetch_ongoing = 0
self.database_engine = hs.database_engine
diff --git a/synapse/storage/events.py b/synapse/storage/events.py
index 0859518b1b..a6b2e7677f 100644
--- a/synapse/storage/events.py
+++ b/synapse/storage/events.py
@@ -506,41 +506,39 @@ class EventsStore(SQLBaseStore):
def do_fetch(txn):
event_list = []
- try:
- with self._event_fetch_lock:
- event_list = self._event_fetch_list
- self._event_fetch_list = []
-
- if not event_list:
- return
-
- event_id_lists = zip(*event_list)[0]
- event_ids = [
- item for sublist in event_id_lists for item in sublist
- ]
- rows = self._fetch_event_rows(txn, event_ids)
-
- row_dict = {
- r["event_id"]: r
- for r in rows
- }
+ while True:
+ try:
+ with self._event_fetch_lock:
+ event_list = self._event_fetch_list
+ self._event_fetch_list = []
+
+ if not event_list:
+ return
+
+ event_id_lists = zip(*event_list)[0]
+ event_ids = [
+ item for sublist in event_id_lists for item in sublist
+ ]
+ rows = self._fetch_event_rows(txn, event_ids)
+
+ row_dict = {
+ r["event_id"]: r
+ for r in rows
+ }
- for ids, d in event_list:
- d.callback(
- [
- row_dict[i] for i in ids
- if i in row_dict
- ]
- )
- except Exception as e:
- for _, d in event_list:
- try:
- reactor.callFromThread(d.errback, e)
- except:
- pass
- finally:
- with self._event_fetch_lock:
- self._event_fetch_ongoing = False
+ for ids, d in event_list:
+ d.callback(
+ [
+ row_dict[i] for i in ids
+ if i in row_dict
+ ]
+ )
+ except Exception as e:
+ for _, d in event_list:
+ try:
+ reactor.callFromThread(d.errback, e)
+ except:
+ pass
def cb(rows):
return defer.gatherResults([
@@ -561,12 +559,12 @@ class EventsStore(SQLBaseStore):
(events, d)
)
- if not self._event_fetch_ongoing:
+ if self._event_fetch_ongoing < 3:
+ self._event_fetch_ongoing += 1
self.runInteraction(
"do_fetch",
do_fetch
)
- self._event_fetch_ongoing = True
res = yield d
|
