diff --git a/cassandra/cluster.py b/cassandra/cluster.py index da1c883cbd..8adb60bb0c 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -577,6 +577,8 @@ def on_down(self, host): return for p in self.profiles.values(): p.load_balancing_policy.on_down(host) + if host._is_removed: + return def on_add(self, host): for p in self.profiles.values(): @@ -2189,6 +2191,10 @@ def on_up(self, host): log.debug("Waiting to acquire lock for handling up status of node %s", host) with host.lock: + if host._is_removed: + log.debug("Ignoring up status for removed host %s", host) + return + if host._currently_handling_node_up: log.debug("Another thread is already handling up status of node %s", host) return @@ -2314,40 +2320,46 @@ def on_down_potentially_blocking(self, host, is_host_addition, down_event_generation, on_add_reconnection=None): try: - with host.lock: - if down_event_generation != host._down_event_generation: - return - # The load balancing policies go first, so that the query plan the # control connection's reconnect walks no longer offers the host # that just went down; otherwise the reconnect submitted below can # start before the policies have dropped it and burn a - # connect_timeout on a dead node. Both calls are guarded because + # connect_timeout on a dead node. Callbacks are guarded because # on_down() reports successful dispatch to callers that rely on a # reconnection being started. try: - self.profile_manager.on_down(host) + if not self._dispatch_down_callback( + host, down_event_generation, + self.profile_manager.on_down): + return except Exception: log.exception("Error in load balancing policy down handler for host %s", host) try: - self.control_connection.on_down(host) + if not self._dispatch_down_callback( + host, down_event_generation, + self.control_connection.on_down): + return except Exception: log.exception("Error in control connection down handler for host %s", host) for session in tuple(self.sessions): try: - session.on_down(host) + if not self._dispatch_down_callback( + host, down_event_generation, session.on_down): + return except Exception: log.exception("Error marking host %s down in session", host) for listener in self.listeners: try: - listener.on_down(host) + if not self._dispatch_down_callback( + host, down_event_generation, listener.on_down): + return except Exception: log.exception("Error in host state listener down handler for host %s", host) # Pool cleanup and listener callbacks above may race a newer DOWN - # generation. Pair the final generation check with installation - # so stale work cannot publish a reconnector after invalidation. + # generation or host removal. Pair the final check with handler + # installation so stale work cannot publish a reconnector. with host.lock: if down_event_generation != host._down_event_generation: return @@ -2362,6 +2374,20 @@ def on_down_potentially_blocking(self, host, is_host_addition, if down_event_generation == host._down_event_generation: host._currently_handling_node_down = False + @staticmethod + def _dispatch_down_callback(host, down_event_generation, callback): + """Invoke a current DOWN callback before removal can overtake it.""" + # Removal waits for this callback before notifying on_remove. User + # callbacks must not wait on cluster executor work; #382 tracks moving + # lifecycle ordering to queued work instead of blocking a worker. + with host._down_callbacks_lock: + with host.lock: + if (host._is_removed or + down_event_generation != host._down_event_generation): + return False + callback(host) + return True + @run_in_executor def _restart_reconnector(self, host, is_host_addition, down_event_generation, @@ -2400,6 +2426,9 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False, restart_reconnector = False with host.lock: + if host._is_removed: + return False + was_up = host.is_up # ignore down signals if we have open pools to the host @@ -2648,13 +2677,14 @@ def on_remove(self, host, trigger_reconciliation=True): log.debug("[cluster] Removing host %s", host) # A Host object is never re-added after lifecycle removal; a later # discovery creates a new object, even when it carries the same ID. - # Mark this instance before removing its pools so pool creation - # already in flight cannot publish after removal has passed it. - with host.lock: - host._is_removed = True - host.set_down() - host._pending_host_addition = False - host._pending_host_addition_callback = None + # Serialize the terminal transition with DOWN callbacks that already + # validated this Host, then mark it before removing its pools. + with host._down_callbacks_lock: + with host.lock: + host._is_removed = True + host.set_down() + host._pending_host_addition = False + host._pending_host_addition_callback = None self.profile_manager.on_remove(host) for session in tuple(self.sessions): session.on_remove( diff --git a/cassandra/policies.py b/cassandra/policies.py index 812603616f..61db69d6f6 100644 --- a/cassandra/policies.py +++ b/cassandra/policies.py @@ -87,7 +87,13 @@ def on_up(self, host): raise NotImplementedError() def on_down(self, host): - """ Called when a node is marked down. """ + """ + Called when a node is marked down. + + Return promptly. Host removal waits for an in-progress DOWN callback, + so do not wait synchronously for work submitted to the cluster executor. + This applies to registered listeners and custom load-balancing policies. + """ raise NotImplementedError() def on_add(self, host): diff --git a/cassandra/pool.py b/cassandra/pool.py index 26b5670e09..632a6702de 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -162,6 +162,7 @@ class Host(object): _down_event_generation = 0 _pending_host_addition = False _pending_host_addition_callback = None + _down_callbacks_lock = None sharding_info = None @@ -182,6 +183,10 @@ def __init__(self, endpoint, conviction_policy_factory, datacenter=None, rack=No self.conviction_policy = conviction_policy_factory(self) self.set_location_info(datacenter, rack) self.lock = RLock() + # DOWN callbacks call into policies and sessions. Keep them ordered + # with terminal removal without holding the Host state lock across + # those external callbacks. + self._down_callbacks_lock = RLock() @property def address(self): diff --git a/tests/unit/test_cluster.py b/tests/unit/test_cluster.py index 11b3e9901b..1e701d87c5 100644 --- a/tests/unit/test_cluster.py +++ b/tests/unit/test_cluster.py @@ -381,6 +381,38 @@ def test_on_up_waits_until_every_session_pool_is_scheduled(self): first_session.remove_pool.assert_has_calls([call(host), call(host)]) second_session.remove_pool.assert_has_calls([call(host), call(host)]) + def test_on_up_ignores_removed_host(self): + cluster = Cluster() + self.addCleanup(cluster.shutdown) + cluster.profile_manager = Mock() + cluster.profile_manager.distance.return_value = HostDistance.LOCAL + cluster.control_connection.on_up = Mock() + cluster.control_connection.on_remove = Mock() + cluster._prepare_all_queries = Mock() + + host = Host( + "127.0.0.1", SimpleConvictionPolicy, host_id=uuid.uuid4()) + cluster.metadata.add_or_return_host(host) + host.set_down() + cluster.remove_host(host) + + session = Mock() + session.add_or_renew_pool = Mock() + session.update_created_pools = Mock() + cluster.sessions = (session,) + listener = Mock() + cluster.register_listener(listener) + + cluster.on_up(host) + + session.remove_pool.assert_not_called() + session.add_or_renew_pool.assert_not_called() + cluster._prepare_all_queries.assert_not_called() + cluster.profile_manager.on_up.assert_not_called() + cluster.control_connection.on_up.assert_not_called() + listener.on_up.assert_not_called() + assert not host._currently_handling_node_up + def test_failed_replacement_add_removes_partial_pools_and_reconnects(self): cluster = Cluster() self.addCleanup(cluster.shutdown) @@ -1451,6 +1483,26 @@ def test_queued_reconnector_restart_skips_recovered_host(self): cluster._start_reconnector.assert_not_called() assert not self.host._currently_handling_node_down + def test_queued_reconnector_restart_skips_removed_host(self): + cluster = Cluster() + cluster.executor.shutdown() + cluster.executor = Mock() + self.addCleanup(cluster.shutdown) + cluster._discount_down_events = False + cluster.metadata.add_or_return_host(self.host) + cluster.executor.submit.return_value = Future() + self.host.set_down() + + assert not cluster.on_down(self.host, is_host_addition=False) + restart_task, *args = cluster.executor.submit.call_args.args + cluster.remove_host(self.host) + + restart_task(*args) + + assert self.host._is_removed + assert not self.host.is_currently_reconnecting() + assert not self.host._currently_handling_node_down + def test_fresh_down_supersedes_queued_reconnector_restart(self): cluster = Cluster() cluster.executor.shutdown() @@ -1707,6 +1759,97 @@ def test_failed_down_submission_releases_pending_state(self): assert not self.host._currently_handling_node_down + def test_queued_down_processing_survives_stale_metadata_index(self): + cluster = Cluster() + cluster.scheduler.shutdown() + cluster.scheduler = Mock() + cluster.executor.shutdown() + cluster.executor = Mock() + self.addCleanup(cluster.shutdown) + cluster._discount_down_events = False + cluster.metadata.add_or_return_host(self.host) + cluster.profile_manager.distance = Mock( + return_value=HostDistance.LOCAL) + cluster.profile_manager.on_down = Mock() + cluster.control_connection.on_down = Mock() + cluster._make_connection_factory = Mock(return_value=Mock()) + session = Mock() + cluster.sessions = (session,) + listener = Mock() + cluster.register_listener(listener) + cluster.executor.submit.return_value = Future() + self.host.set_up() + + assert cluster.on_down(self.host, is_host_addition=False) + down_task, *args = cluster.executor.submit.call_args.args + current_host_id = self.host.host_id + stale_host_id = uuid.uuid4() + # Model the transient middle of a metadata reindex without mutating + # the Host's immutable identity. + with cluster.metadata._hosts_lock: + popped_host = cluster.metadata._hosts.pop(current_host_id) + assert popped_host is self.host + cluster.metadata._hosts[stale_host_id] = self.host + cluster.metadata._host_id_by_endpoint[ + self.host.endpoint] = stale_host_id + + assert self.host.host_id == current_host_id + assert cluster.metadata.get_host_by_host_id(current_host_id) is None + assert cluster.metadata.get_host(self.host.endpoint) is self.host + assert cluster.metadata.all_hosts_items() == [ + (stale_host_id, self.host)] + assert not self.host._is_removed + + down_task(*args) + + cluster.profile_manager.on_down.assert_called_once_with(self.host) + cluster.control_connection.on_down.assert_called_once_with(self.host) + session.on_down.assert_called_once_with(self.host) + listener.on_down.assert_called_once_with(self.host) + cluster._make_connection_factory.assert_called_once_with(self.host) + assert self.host.is_currently_reconnecting() + cluster.scheduler.schedule.assert_called_once() + assert not self.host._currently_handling_node_down + + def test_removed_host_rejects_queued_and_new_down_processing(self): + cluster = Cluster() + cluster.scheduler.shutdown() + cluster.scheduler = Mock() + cluster.executor.shutdown() + cluster.executor = Mock() + self.addCleanup(cluster.shutdown) + cluster._discount_down_events = False + cluster.metadata.add_or_return_host(self.host) + cluster.profile_manager.on_down = Mock() + cluster.control_connection.on_down = Mock() + cluster._start_reconnector = Mock() + session = Mock() + cluster.sessions = (session,) + listener = Mock() + cluster.register_listener(listener) + pending_future = Future() + cluster.executor.submit.return_value = pending_future + self.host.set_up() + + assert cluster.on_down(self.host, is_host_addition=False) + down_task, *args = cluster.executor.submit.call_args.args + cluster.remove_host(self.host) + cluster.profile_manager.on_down.reset_mock() + cluster.control_connection.on_down.reset_mock() + + down_task(*args) + assert not cluster.on_down(self.host, is_host_addition=False) + + cluster.profile_manager.on_down.assert_not_called() + cluster.control_connection.on_down.assert_not_called() + session.on_down.assert_not_called() + listener.on_down.assert_not_called() + cluster._start_reconnector.assert_not_called() + cluster.executor.submit.assert_called_once() + assert self.host._is_removed + assert not self.host._currently_handling_node_down + + class SchedulerTest(unittest.TestCase): # TODO: this suite could be expanded; for now just adding a test covering a ticket @@ -2770,6 +2913,120 @@ def test_down_handling_continues_when_a_listener_raises(self): self.cluster._start_reconnector.assert_called_once_with( self.host, False) + def test_reentrant_removal_stops_remaining_down_callbacks(self): + del self.cluster.on_down_potentially_blocking # use the real method + self.cluster.metadata.add_or_return_host(self.host) + self.cluster.control_connection = Mock() + session = Mock() + self.cluster.sessions = (session,) + listener = Mock() + self.cluster.register_listener(listener) + self.cluster._start_reconnector = Mock() + self.cluster.profile_manager.on_down = Mock( + side_effect=lambda host: self.cluster.remove_host(host)) + body = Cluster.on_down_potentially_blocking.__wrapped__ + + body(self.cluster, self.host, is_host_addition=False, + down_event_generation=self.host._down_event_generation) + + assert self.host._is_removed + self.cluster.control_connection.on_down.assert_not_called() + session.on_down.assert_not_called() + listener.on_down.assert_not_called() + self.cluster._start_reconnector.assert_not_called() + + def test_reentrant_policy_removal_stops_later_policy_callbacks(self): + del self.cluster.on_down_potentially_blocking # use the real method + self.cluster.metadata.add_or_return_host(self.host) + removing_policy = Mock() + following_policy = Mock() + removing_policy.on_down.side_effect = \ + lambda host: self.cluster.remove_host(host) + self.cluster.profile_manager.profiles = { + 'removing': ExecutionProfile(removing_policy), + 'following': ExecutionProfile(following_policy), + } + self.cluster.control_connection = Mock() + self.cluster._start_reconnector = Mock() + body = Cluster.on_down_potentially_blocking.__wrapped__ + + body(self.cluster, self.host, is_host_addition=False, + down_event_generation=self.host._down_event_generation) + + assert self.host._is_removed + removing_policy.on_down.assert_called_once_with(self.host) + following_policy.on_down.assert_not_called() + self.cluster.control_connection.on_down.assert_not_called() + self.cluster._start_reconnector.assert_not_called() + + def test_remove_waits_for_inflight_down_callback(self): + del self.cluster.on_down_potentially_blocking # use the real method + self.cluster.metadata.add_or_return_host(self.host) + self.cluster.profile_manager.on_down = Mock() + self.cluster._start_reconnector = Mock() + callback_started = Event() + release_callback = Event() + removal_attempted = Event() + removal_finished = Event() + callback_order = [] + + def on_control_down(_host): + callback_started.set() + assert release_callback.wait(5) + callback_order.append('down') + + self.cluster.control_connection = Mock() + self.cluster.control_connection.on_down.side_effect = on_control_down + listener = Mock() + listener.on_remove.side_effect = \ + lambda _host: callback_order.append('remove') + self.cluster.register_listener(listener) + self.host.set_down() + self.host._currently_handling_node_down = True + + callback_lock = self.host._down_callbacks_lock + remove_thread = None + + class AttemptTrackingLock(object): + + def __enter__(self): + if threading.current_thread() is remove_thread: + removal_attempted.set() + callback_lock.acquire() + return self + + def __exit__(self, exc_type, exc_value, traceback): + callback_lock.release() + + self.host._down_callbacks_lock = AttemptTrackingLock() + body = Cluster.on_down_potentially_blocking.__wrapped__ + down_thread = Thread( + target=body, + args=(self.cluster, self.host, False, + self.host._down_event_generation)) + + def remove_host(): + try: + self.cluster.remove_host(self.host) + finally: + removal_finished.set() + + remove_thread = Thread(target=remove_host) + down_thread.start() + assert callback_started.wait(5) + remove_thread.start() + assert removal_attempted.wait(5) + removal_was_blocked = not removal_finished.is_set() + release_callback.set() + down_thread.join(5) + remove_thread.join(5) + + assert removal_was_blocked + assert not down_thread.is_alive() + assert not remove_thread.is_alive() + assert callback_order.index('down') < callback_order.index('remove') + assert self.host._is_removed + def test_signal_connection_failure_rejected_by_conviction_policy(self): error = ConnectionException("connection failed") self.host.signal_connection_failure = Mock(return_value=False)