Fix the race on the timers

Formerly, we were doing a logic like the following: Check the
timer in front of the priority queue of timers. If it is expired,
call the callback and pop it from the queue. However, this logic
is flawed because the timers queue was accessed from multiple threads.
So, it may be the case that, we check the timer in front of the queue,
run its callback, then, some other thread adds another timer to the
queue concurrently and it is put in front of the queue. Now, when the
reactor thread pops the item in front of the queue, it will be the newly
added timer, before its callback being executed.

To solve this, we use a double buffering approach. A thead-safe queue
is used to store newly added timers which can be efficiently modified
on both ends on multiple threads. Then, the reactor thread periodically
pops items from that queue in FIFO order and maintains a min heap
using a list with the help of heappush and heappop functions. Only if
the min heap has some elements, timers are popped from that and executed.

Also, the unneeded `timer_cancelled_cb` is field is removed. We were
removing the timer from the queue after canceling it. That was the
only usage of this field. We do the same with the new approach,
if it is canceled, it will return `True` on `check_timer` call without
executing its `timer_ended_cb` and will be removed from the heap.

Author: mdumandag

hazelcast/reactor.py

@@ -9,14 +9,14 @@
 
 from collections import deque
 from functools import total_ordering
+from heapq import heappush, heappop
 
 from hazelcast import six
 from hazelcast.config import SSLProtocol
 from hazelcast.connection import Connection
 from hazelcast.core import Address
 from hazelcast.errors import HazelcastError
 from hazelcast.future import Future
-from hazelcast.six.moves import queue
 
 try:
     import ssl
@@ -31,7 +31,8 @@ class AsyncoreReactor(object):
     _is_live = False
 
     def __init__(self):
-        self._timers = queue.PriorityQueue()
+        self._timers = []  # Accessed only from the reactor thread
+        self._new_timers = deque()  # Popped only from the reactor thread
         self._map = {}
 
     def start(self):
@@ -55,28 +56,36 @@ def _loop(self):
                 _logger.exception("Error in Reactor Thread")
                 # TODO: shutdown client
                 return
-        _logger.debug("Reactor Thread exited. %s" % self._timers.qsize())
+        _logger.debug("Reactor Thread exited")
         self._cleanup_all_timers()
 
     def _check_timers(self):
-        now = time.time()
-        while not self._timers.empty():
-            try:
-                timer = self._timers.queue[0][1]
-            except IndexError:
-                return
-
-            if timer.check_timer(now):
-                try:
-                    self._timers.get_nowait()
-                except queue.Empty:
-                    pass
-            else:
-                return
+        timers = self._timers
+
+        if self._new_timers:
+            new_timers = self._new_timers
+            while new_timers:
+                # There is no need to check for exception here,
+                # reactor thread is the only one popping from
+                # the deque. So, if the we pass the size check
+                # above, there should be at least one element
+                heappush(timers, new_timers.popleft())
+
+        if timers:
+            now = time.time()
+            while timers:
+                timer = timers[0][1]
+                if timer.check_timer(now):
+                    heappop(timers)
+                else:
+                    # Timer in the root of the min heap is not expired
+                    # Therefore, there should not be any expired
+                    # timers in the heap.
+                    return
 
     def add_timer_absolute(self, timeout, callback):
-        timer = Timer(timeout, callback, self._cleanup_timer)
-        self._timers.put_nowait((timer.end, timer))
+        timer = Timer(timeout, callback)
+        self._new_timers.append((timer.end, timer))
         return timer
 
     def add_timer(self, delay, callback):
@@ -105,19 +114,18 @@ def connection_factory(self, connection_manager, connection_id, address, network
         return AsyncoreConnection(self._map, connection_manager, connection_id,
                                   address, network_config, message_callback)
 
-    def _cleanup_timer(self, timer):
-        try:
-            self._timers.queue.remove((timer.end, timer))
-        except ValueError:
-            pass
-
     def _cleanup_all_timers(self):
-        while not self._timers.empty():
-            try:
-                _, timer = self._timers.get_nowait()
-                timer.timer_ended_cb()
-            except queue.Empty:
-                return
+        self._check_timers()
+
+        timers = self._timers
+        while timers:
+            _, timer = timers.pop()
+            timer.timer_ended_cb()
+
+        # Although it is not the case with the current code base,
+        # the timers ended above may add new timers.
+        if self._new_timers:
+            self._cleanup_all_timers()
 
 
 _BUFFER_SIZE = 128000
@@ -274,26 +282,24 @@ def __str__(self):
 
 @total_ordering
 class Timer(object):
-    __slots__ = ("end", "timer_ended_cb", "timer_canceled_cb", "canceled")
+    __slots__ = ("end", "timer_ended_cb", "canceled")
 
-    def __init__(self, end, timer_ended_cb, timer_canceled_cb):
+    def __init__(self, end, timer_ended_cb):
         self.end = end
         self.timer_ended_cb = timer_ended_cb
-        self.timer_canceled_cb = timer_canceled_cb
         self.canceled = False
 
     def __eq__(self, other):
         return self.end == other.end
 
     def __ne__(self, other):
-        return not (self == other)
+        return self.end != other.end
 
     def __lt__(self, other):
         return self.end < other.end
 
     def cancel(self):
         self.canceled = True
-        self.timer_canceled_cb(self)
 
     def check_timer(self, now):
         if self.canceled: