Add a wait

synapse/storage/_base.py

@@ -301,7 +301,7 @@ class SQLBaseStore(object):
         self._get_event_cache = Cache("*getEvent*", keylen=3, lru=True,
                                       max_entries=hs.config.event_cache_size)
 
-        self._event_fetch_lock = threading.Lock()
+        self._event_fetch_lock = threading.Condition()
         self._event_fetch_list = []
         self._event_fetch_ongoing = 0
 

synapse/storage/events.py

@@ -502,6 +502,7 @@ class EventsStore(SQLBaseStore):
 
     def _do_fetch(self, conn):
         event_list = []
+        i = 0
         while True:
             try:
                 logger.debug("do_fetch getting lock")
@@ -510,8 +511,14 @@ class EventsStore(SQLBaseStore):
                     event_list = self._event_fetch_list
                     self._event_fetch_list = []
                     if not event_list:
-                        self._event_fetch_ongoing -= 1
-                        return
+                        if self.database_engine.single_threaded or i > 5:
+                            self._event_fetch_ongoing -= 1
+                            return
+                        else:
+                            self._event_fetch_lock.wait(0.1)
+                            i += 1
+                            continue
+                    i = 0
 
                 event_id_lists = zip(*event_list)[0]
                 event_ids = [
@@ -566,6 +573,8 @@ class EventsStore(SQLBaseStore):
                 (events, events_d)
             )
 
+            self._event_fetch_lock.notify_all()
+
             if self._event_fetch_ongoing < 1:
                 self._event_fetch_ongoing += 1
                 should_start = True