summary refs log tree commit diff
path: root/synapse/notifier.py
diff options
context:
space:
mode:
authorErik Johnston <erik@matrix.org>2016-12-09 15:43:18 +0000
committerErik Johnston <erik@matrix.org>2016-12-09 15:43:18 +0000
commitbe14c24cea7d96f850180759870e83c3789bcdda (patch)
tree69af80fd2504959b1dcf811279adff26b2f77737 /synapse/notifier.py
parentMerge pull request #1680 from matrix-org/erikj/joined_rooms (diff)
downloadsynapse-be14c24cea7d96f850180759870e83c3789bcdda.tar.xz
Fix rare notifier bug where listeners dont timeout
There was a race condition that caused the notifier to 'miss' the
timeout notification, since there were no other checks for the timeout
this caused listeners to get stuck in a loop until something happened.
Diffstat (limited to 'synapse/notifier.py')
-rw-r--r--synapse/notifier.py20
1 files changed, 9 insertions, 11 deletions
diff --git a/synapse/notifier.py b/synapse/notifier.py
index 054ca59ad2..40baa6969a 100644
--- a/synapse/notifier.py
+++ b/synapse/notifier.py
@@ -294,14 +294,7 @@ class Notifier(object):
 
         result = None
         if timeout:
-            # Will be set to a _NotificationListener that we'll be waiting on.
-            # Allows us to cancel it.
-            listener = None
-
-            def timed_out():
-                if listener:
-                    listener.deferred.cancel()
-            timer = self.clock.call_later(timeout / 1000., timed_out)
+            end_time = self.clock.time_msec() + timeout
 
             prev_token = from_token
             while not result:
@@ -312,6 +305,10 @@ class Notifier(object):
                     if result:
                         break
 
+                    now = self.clock.time_msec()
+                    if end_time <= now:
+                        break
+
                     # Now we wait for the _NotifierUserStream to be told there
                     # is a new token.
                     # We need to supply the token we supplied to callback so
@@ -319,11 +316,12 @@ class Notifier(object):
                     prev_token = current_token
                     listener = user_stream.new_listener(prev_token)
                     with PreserveLoggingContext():
-                        yield listener.deferred
+                        yield self.clock.time_bound_deferred(
+                            listener.deferred,
+                            time_out=(end_time - now) / 1000.
+                        )
                 except defer.CancelledError:
                     break
-
-            self.clock.cancel_call_later(timer, ignore_errs=True)
         else:
             current_token = user_stream.current_token
             result = yield callback(from_token, current_token)