cluster: replay queued up events

Author: dkropachev

cassandra/cluster.py

@@ -234,6 +234,7 @@ def new_f(self, *args, **kwargs):
         try:
             future = self.executor.submit(f, self, *args, **kwargs)
             future.add_done_callback(_future_completed)
+            return future
         except Exception:
             log.exception("Failed to submit task to executor")
 
@@ -1870,7 +1871,61 @@ def _cleanup_failed_on_up_handling(self, host):
 
         self._start_reconnector(host, is_host_addition=False)
 
-    def _on_up_future_completed(self, host, futures, results, lock, finished_future):
+    def _up_handling_was_superseded(self, host, up_handling_revision):
+        return (getattr(host, "_node_up_handling_revision", None) != up_handling_revision or
+                host._node_down_event_revision != up_handling_revision)
+
+    def _clear_up_handling(self, host, up_handling_revision=None):
+        if (up_handling_revision is not None and
+                getattr(host, "_node_up_handling_revision", None) != up_handling_revision):
+            return False
+        host._currently_handling_node_up = False
+        host._node_up_handling_revision = None
+        return True
+
+    def _cleanup_superseded_up_handling(self, host):
+        for session in tuple(self.sessions):
+            session.remove_pool(host)
+
+    def _pop_pending_node_up_if_ready(self, host):
+        if not host._pending_node_up:
+            return None
+        if host.is_up:
+            host._pending_node_up = False
+            host._pending_node_up_revision = None
+            return None
+        if host._currently_handling_node_up or host._currently_handling_node_down:
+            return None
+
+        pending_up_revision = host._pending_node_up_revision
+        # Leave the pending marker in place until on_up() reacquires host.lock so
+        # a newer down signal can still invalidate this replay.
+        return pending_up_revision
+
+    def _handle_pending_node_up(self, host, pending_up_revision):
+        if pending_up_revision is not None:
+            log.debug("Handling queued up status of node %s", host)
+            self.on_up(host, expected_down_event_revision=pending_up_revision)
+
+    def _clear_down_handling(self, host, down_event_revision=None):
+        if (down_event_revision is not None and
+                getattr(host, "_node_down_handling_revision", None) != down_event_revision):
+            return False
+        host._currently_handling_node_down = False
+        host._node_down_handling_revision = None
+        return True
+
+    def _finish_superseded_up_handling(self, host, up_handling_revision):
+        self._cleanup_superseded_up_handling(host)
+
+        pending_up_revision = None
+        with host.lock:
+            if self._clear_up_handling(host, up_handling_revision):
+                pending_up_revision = self._pop_pending_node_up_if_ready(host)
+
+        self._handle_pending_node_up(host, pending_up_revision)
+
+    def _on_up_future_completed(self, host, up_handling_revision, futures, results, lock, finished_future):
         with lock:
             futures.discard(finished_future)
 
@@ -1896,18 +1951,31 @@ def _on_up_future_completed(self, host, futures, results, lock, finished_future)
 
             log.info("Connection pools established for node %s", host)
             # mark the host as up and notify all listeners
-            host.set_up()
+            superseded = False
+            with host.lock:
+                if self._up_handling_was_superseded(host, up_handling_revision):
+                    log.debug("Ignoring superseded up handling for node %s", host)
+                    superseded = True
+                else:
+                    host.set_up()
+                    self._clear_up_handling(host, up_handling_revision)
+            if superseded:
+                self._finish_superseded_up_handling(host, up_handling_revision)
+                return
             for listener in self.listeners:
                 listener.on_up(host)
         finally:
+            pending_up_revision = None
             with host.lock:
-                host._currently_handling_node_up = False
+                if self._clear_up_handling(host, up_handling_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+            self._handle_pending_node_up(host, pending_up_revision)
 
         # see if there are any pools to add or remove now that the host is marked up
         for session in tuple(self.sessions):
             session.update_created_pools()
 
-    def on_up(self, host):
+    def on_up(self, host, expected_down_event_revision=None):
         """
         Intended for internal use only.
         """
@@ -1916,14 +1984,38 @@ def on_up(self, host):
 
         log.debug("Waiting to acquire lock for handling up status of node %s", host)
         with host.lock:
+            if (expected_down_event_revision is not None and
+                    host._node_down_event_revision != expected_down_event_revision):
+                log.debug("Ignoring stale queued up handling for node %s", host)
+                return
+
+            if host._currently_handling_node_down:
+                log.debug("Down status is being handled for node %s; queueing up handling", host)
+                host._pending_node_up = True
+                host._pending_node_up_revision = host._node_down_event_revision
+                return
+
             if host._currently_handling_node_up:
-                log.debug("Another thread is already handling up status of node %s", host)
+                up_handling_revision = getattr(host, "_node_up_handling_revision", None)
+                if self._up_handling_was_superseded(host, up_handling_revision):
+                    log.debug("Superseded up handling is still finishing for node %s; "
+                              "queueing up handling", host)
+                    host._pending_node_up = True
+                    host._pending_node_up_revision = host._node_down_event_revision
+                else:
+                    log.debug("Another thread is already handling up status of node %s", host)
                 return
 
             if host.is_up:
                 log.debug("Host %s was already marked up", host)
+                host._pending_node_up = False
+                host._pending_node_up_revision = None
                 return
 
+            host._pending_node_up = False
+            host._pending_node_up_revision = None
+            up_handling_revision = host._node_down_event_revision
+            host._node_up_handling_revision = up_handling_revision
             host._currently_handling_node_up = True
         log.debug("Starting to handle up status of node %s", host)
 
@@ -1953,7 +2045,8 @@ def on_up(self, host):
             log.debug("Attempting to open new connection pools for host %s", host)
             futures_lock = Lock()
             futures_results = []
-            callback = partial(self._on_up_future_completed, host, futures, futures_results, futures_lock)
+            callback = partial(self._on_up_future_completed, host, up_handling_revision,
+                               futures, futures_results, futures_lock)
             for session in tuple(self.sessions):
                 future = session.add_or_renew_pool(host, is_host_addition=False)
                 if future is not None:
@@ -1967,14 +2060,24 @@ def on_up(self, host):
 
             self._cleanup_failed_on_up_handling(host)
 
+            pending_up_revision = None
             with host.lock:
-                host._currently_handling_node_up = False
+                if self._clear_up_handling(host, up_handling_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+            self._handle_pending_node_up(host, pending_up_revision)
             raise
         else:
             if not have_future:
+                superseded = False
                 with host.lock:
-                    host.set_up()
-                    host._currently_handling_node_up = False
+                    if self._up_handling_was_superseded(host, up_handling_revision):
+                        log.debug("Ignoring superseded up handling for node %s", host)
+                        superseded = True
+                    else:
+                        host.set_up()
+                        self._clear_up_handling(host, up_handling_revision)
+                if superseded:
+                    self._finish_superseded_up_handling(host, up_handling_revision)
 
         # for testing purposes
         return futures
@@ -2004,16 +2107,59 @@ def _start_reconnector(self, host, is_host_addition):
         reconnector.start()
 
     @run_in_executor
-    def on_down_potentially_blocking(self, host, is_host_addition):
-        self.profile_manager.on_down(host)
-        self.control_connection.on_down(host)
-        for session in tuple(self.sessions):
-            session.on_down(host)
+    def on_down_potentially_blocking(self, host, is_host_addition, down_revision=None, down_event_revision=None):
+        handle_stale_down_event = False
+        pending_up_revision = None
+        with host.lock:
+            down_handling_revision = getattr(host, "_node_down_handling_revision", None)
+            owns_reserved_down_handling = (
+                down_event_revision is not None and
+                host._currently_handling_node_down and
+                down_handling_revision == down_event_revision
+            )
+            node_up_handling_revision = getattr(host, "_node_up_handling_revision", None)
+            stale_down_event = (
+                host.is_up or
+                (host._currently_handling_node_up and
+                 (down_event_revision is None or
+                  node_up_handling_revision is None or
+                  down_event_revision <= node_up_handling_revision)) or
+                (down_revision is not None and host._state_revision != down_revision)
+            )
+            if stale_down_event:
+                log.debug("Ignoring stale down handling for host %s", host)
+                if owns_reserved_down_handling and self._clear_down_handling(host, down_event_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+                handle_stale_down_event = True
+            elif host._currently_handling_node_down:
+                if not owns_reserved_down_handling:
+                    log.debug("Another thread is already handling down status of node %s", host)
+                    return
+            else:
+                host._currently_handling_node_down = True
+                host._node_down_handling_revision = down_event_revision
 
-        for listener in self.listeners:
-            listener.on_down(host)
+        if handle_stale_down_event:
+            self._handle_pending_node_up(host, pending_up_revision)
+            return
 
-        self._start_reconnector(host, is_host_addition)
+        try:
+            self.profile_manager.on_down(host)
+            self.control_connection.on_down(host)
+            for session in tuple(self.sessions):
+                session.on_down(host)
+
+            for listener in self.listeners:
+                listener.on_down(host)
+
+            self._start_reconnector(host, is_host_addition)
+        finally:
+            pending_up_revision = None
+            with host.lock:
+                if self._clear_down_handling(host, down_event_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+
+            self._handle_pending_node_up(host, pending_up_revision)
 
     def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
         """
@@ -2037,12 +2183,39 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False):
                 if connected:
                     return
 
+            if not expect_host_to_be_down:
+                if was_up is False:
+                    if host._pending_node_up:
+                        host._node_down_event_revision += 1
+                        host._pending_node_up = False
+                        host._pending_node_up_revision = None
+                        host.set_down()
+                    return
+                if was_up is None:
+                    return
+
+            host._node_down_event_revision += 1
+            host._pending_node_up = False
+            host._pending_node_up_revision = None
             host.set_down()
-            if (not was_up and not expect_host_to_be_down) or host.is_currently_reconnecting():
+            down_revision = host._state_revision
+            down_event_revision = host._node_down_event_revision
+            if host.is_currently_reconnecting():
+                return
+            if host._currently_handling_node_down:
                 return
+            host._currently_handling_node_down = True
+            host._node_down_handling_revision = down_event_revision
         log.warning("Host %s has been marked down", host)
 
-        self.on_down_potentially_blocking(host, is_host_addition)
+        future = self.on_down_potentially_blocking(
+            host, is_host_addition, down_revision, down_event_revision)
+        if future is None:
+            pending_up_revision = None
+            with host.lock:
+                if self._clear_down_handling(host, down_event_revision):
+                    pending_up_revision = self._pop_pending_node_up_if_ready(host)
+            self._handle_pending_node_up(host, pending_up_revision)
 
     def on_add(self, host, refresh_nodes=True):
         if self.is_shutdown:
@@ -2119,7 +2292,11 @@ def on_remove(self, host):
             return
 
         log.debug("[cluster] Removing host %s", host)
-        host.set_down()
+        with host.lock:
+            host._node_down_event_revision += 1
+            host._pending_node_up = False
+            host._pending_node_up_revision = None
+            host.set_down()
         self.profile_manager.on_remove(host)
         for session in tuple(self.sessions):
             session.on_remove(host)

cassandra/pool.py

@@ -164,6 +164,12 @@ class Host(object):
     lock = None
 
     _currently_handling_node_up = False
+    _currently_handling_node_down = False
+    _node_down_handling_revision = None
+    _pending_node_up = False
+    _pending_node_up_revision = None
+    _node_down_event_revision = 0
+    _state_revision = 0
 
     sharding_info = None
 
@@ -180,6 +186,13 @@ def __init__(self, endpoint, conviction_policy_factory, datacenter=None, rack=No
         self.host_id = host_id
         self.set_location_info(datacenter, rack)
         self.lock = RLock()
+        self._currently_handling_node_up = False
+        self._currently_handling_node_down = False
+        self._node_down_handling_revision = None
+        self._pending_node_up = False
+        self._pending_node_up_revision = None
+        self._node_down_event_revision = 0
+        self._state_revision = 0
 
     @property
     def address(self):
@@ -212,9 +225,15 @@ def set_up(self):
         if not self.is_up:
             log.debug("Host %s is now marked up", self.endpoint)
         self.conviction_policy.reset()
+        if self.is_up is not True:
+            self._state_revision += 1
         self.is_up = True
+        self._pending_node_up = False
+        self._pending_node_up_revision = None
 
     def set_down(self):
+        if self.is_up is not False:
+            self._state_revision += 1
         self.is_up = False
 
     def signal_connection_failure(self, connection_exc):
@@ -920,5 +939,3 @@ def open_count(self):
     @property
     def _excess_connection_limit(self):
         return self.host.sharding_info.shards_count * self.max_excess_connections_per_shard_multiplier
-
-