diff --git a/cassandra/_lifecycle.py b/cassandra/_lifecycle.py index ccb7910423..e96884a900 100644 --- a/cassandra/_lifecycle.py +++ b/cassandra/_lifecycle.py @@ -20,7 +20,7 @@ from collections import deque from concurrent.futures import Future -from contextlib import contextmanager, nullcontext +from contextlib import contextmanager from functools import partial, wraps import logging from threading import local, RLock @@ -56,9 +56,73 @@ def owns_locked(self): self.host, self.generation, self.expected_pool, self.endpoint)) def is_current(self): + with self.locked(): + return self.owns_locked() + + @contextmanager + def locked(self): with self.session._lock: with self.host.lock: - return self.owns_locked() + yield + + +class ConnectionFailureReservation: + """Retain the pool and shard slot that own a connection failure.""" + + def __init__(self, pool, connection, replacing=False): + self.pool = pool + self.connection = connection + self.session = pool._session_ref() + self.host = pool.host + self.replacing = replacing + self.retired = False + + @contextmanager + def locked(self): + if self.session is None: + yield + return + while True: + with self.session._lock: + with self.host.lock: + # Unpublished candidates can hold their work lock while + # publication waits for Session under the membership lock. + # A completed retirement no longer needs shard ownership + # inspected; its receipt retains only Session/Host guards. + if (self.retired or + self.session._pools.get(self.host) is not self.pool): + yield + return + if self.pool._lock.acquire(False): + try: + yield + finally: + self.pool._lock.release() + return + # Never retain work while reacquiring Session/Host. Active socket + # failures may await I/O under work before they can retire a pool. + with self.pool._lock: + pass + + def owns_locked(self): + if (self.session is None or + self.session._pools.get(self.host) is not self.pool): + return False + owns_slot = self.retired + if not self.retired and not self.pool.is_shutdown: + current = self.pool._connections.get(self.connection.features.shard_id) + owns_slot = (current is self.connection or + (self.replacing and current is None and + self.pool._replacement_connections.get( + self.connection.features.shard_id) is self.connection)) + return (not self.session.is_shutdown and + not self.session.cluster.is_shutdown and + not self.host._is_removed and + self.host.endpoint == self.pool.endpoint and owns_slot) + + def is_current(self): + with self.locked(): + return self.owns_locked() class LifecycleCoordinator: @@ -196,27 +260,26 @@ def observe_health(self, host): def begin_down(self, host, health_generation, connected=False, expect_down=False, pool_failure=None): """Return (generation, retry_only), or None if DOWN is discounted.""" - failure_lock = (pool_failure.session._lock - if pool_failure is not None else nullcontext()) + failure_lock = (pool_failure.locked() + if pool_failure is not None else host.lock) with failure_lock: - with host.lock: - if (self.closed or host._is_removed or connected or - health_generation != host._health_generation or - (pool_failure is not None and not pool_failure.owns_locked())): - return None - host._health_generation += 1 - self.cancel_preparation(host) - host._readiness_generation = ( - getattr(host, '_readiness_generation', 0) + 1) - host._currently_handling_node_up = False - was_up = host.is_up - host.set_down() - if (not was_up and (host.is_currently_reconnecting() or - host._currently_handling_node_down)): - return None - host._currently_handling_node_down = True - host._down_event_generation += 1 - return host._down_event_generation, not was_up and not expect_down + if (self.closed or host._is_removed or connected or + health_generation != host._health_generation or + (pool_failure is not None and not pool_failure.owns_locked())): + return None + host._health_generation += 1 + self.cancel_preparation(host) + host._readiness_generation = ( + getattr(host, '_readiness_generation', 0) + 1) + host._currently_handling_node_up = False + was_up = host.is_up + host.set_down() + if (not was_up and (host.is_currently_reconnecting() or + host._currently_handling_node_down)): + return None + host._currently_handling_node_down = True + host._down_event_generation += 1 + return host._down_event_generation, not was_up and not expect_down def dispatch_down(self, host, health_generation, is_host_addition, connected, expect_host_to_be_down, on_add_reconnection, @@ -242,8 +305,8 @@ def dispatch_down(self, host, health_generation, is_host_addition, on_add_reconnection=on_add_reconnection) if future is None: self.finish_down(host, down_event_generation) - # Restarting the host reconnector does not dispatch the DOWN - # callbacks that reconnect the control connection. + # Restarting the host reconnector does not dispatch + # DOWN callbacks that reconnect the control connection. return False log.warning("Host %s has been marked down", host) diff --git a/cassandra/cluster.py b/cassandra/cluster.py index 1962f95a65..67d933bc88 100644 --- a/cassandra/cluster.py +++ b/cassandra/cluster.py @@ -2655,6 +2655,12 @@ def on_down(self, host, is_host_addition, expect_host_to_be_down=False, if (self._discount_down_events and self.profile_manager.distance(host) != HostDistance.IGNORED): for session in tuple(self.sessions): + # This failure will shut down its pool, including any healthy + # sibling shards. Only independent pools can discount DOWN. + if (isinstance(_pool_failure, + _lifecycle.ConnectionFailureReservation) and + session is _pool_failure.session): + continue pool_state = session.get_pool_state().get(host) if pool_state: connected |= pool_state['open_count'] > 0 diff --git a/cassandra/pool.py b/cassandra/pool.py index a9c35cc375..63a4827fe1 100644 --- a/cassandra/pool.py +++ b/cassandra/pool.py @@ -30,7 +30,7 @@ except ImportError: from cassandra.util import WeakSet # NOQA -from cassandra import AuthenticationFailed +from cassandra import AuthenticationFailed, _lifecycle from cassandra.connection import (ConnectionException, ConnectionSendError, EndPoint, DefaultEndPoint, UnixSocketEndPoint) from cassandra.policies import HostDistance @@ -468,11 +468,12 @@ def __init__(self, host, host_distance, session): self.endpoint = host.endpoint self.host_distance = host_distance self._session = weakref.proxy(session) + self._session_ref = weakref.ref(session) self._lock = Lock() self._membership_lock = Lock() # this is used in conjunction with the connection streams. Not using the connection lock because the connection can be replaced in the lifetime of the pool. self._stream_available_condition = Condition(Lock()) - self._is_replacing = False + self._replacement_connections = {} self._connecting = set() self._connections = {} self._pending_connections = [] @@ -654,41 +655,62 @@ def return_connection(self, connection, stream_was_orphaned=False, self._stream_available_condition.notify() if connection.is_defunct or connection.is_closed: - if connection.signaled_error and not replace_only and \ - (not self.shutdown_on_error or - (isinstance(connection.last_error, ConnectionSendError) and - not heartbeat_failed)): + # Claim the transport before pool work: a redundant reactor + # callback must return while that work waits for its response. + with connection.lock: + if connection.signaled_error and not replace_only and \ + (not self.shutdown_on_error or + (isinstance(connection.last_error, ConnectionSendError) and + not heartbeat_failed)): + return + report_failure = not replace_only and not connection.signaled_error + connection.signaled_error = True + reservation = _lifecycle.ConnectionFailureReservation(self, connection) + with reservation.locked(): + current = reservation.owns_locked() + if not current: + with self._lock: + self._trash.discard(connection) + # A stale completion still releases its stream above, but its + # failed transport cannot affect the current pool or shard. + connection.close() return is_down = False - if replace_only: - # A local send-queue failure says nothing about host health. - # Mark the connection handled so later callbacks do not report - # this failure to the host conviction policy. - connection.signaled_error = True - elif not connection.signaled_error: + if report_failure: log.debug("Defunct or closed connection (%s) returned to pool, potentially " "marking host %s as down", id(connection), self.host) is_down = self.host.signal_connection_failure(connection.last_error) - connection.signaled_error = True if self.shutdown_on_error and not is_down and not replace_only: is_down = True if is_down: - self.shutdown() - self._session.cluster.on_down(self.host, is_host_addition=False) + # Retire before inspecting other sessions: concurrent failures + # must not discount each other's doomed sibling connections. + self.shutdown(_failure=reservation) + if reservation.is_current(): + self._session.cluster.on_down( + self.host, is_host_addition=False, + _pool_failure=reservation) + elif not reservation.retired: + with self._lock: + self._trash.discard(connection) + connection.close() else: + schedule_replacement = False + with reservation.locked(): + if reservation.owns_locked(): + with self._membership_lock: + self._connections.pop(connection.features.shard_id) + self._replacement_connections[ + connection.features.shard_id] = connection + schedule_replacement = True + if not schedule_replacement: + with self._lock: + self._trash.discard(connection) connection.close() - with self._lock: - if self.is_shutdown: - return - with self._membership_lock: - self._connections.pop( - connection.features.shard_id, None) - if self._is_replacing: - return - self._is_replacing = True + if schedule_replacement: self._session.submit(self._replace, connection) elif connection in self._trash: with connection.lock: @@ -712,11 +734,25 @@ def on_orphaned_stream_released(self): self._stream_available_condition.notify() def _replace(self, connection): + reservation = _lifecycle.ConnectionFailureReservation( + self, connection, replacing=True) + if not reservation.is_current(): + with self._lock: + shard_id = connection.features.shard_id + if self._replacement_connections.get(shard_id) is connection: + self._replacement_connections.pop(shard_id) + return failed_connection = None with self._lock: - if self.is_shutdown: + shard_id = connection.features.shard_id + current = self._connections.get(shard_id) + if (self.is_shutdown or + (current is not None and current is not connection) or + (current is None and + self._replacement_connections.get(shard_id) is not connection)): return + self._replacement_connections[shard_id] = connection log.debug("Replacing connection (%s) to %s", id(connection), self.host) try: with self._membership_lock: @@ -725,41 +761,52 @@ def _replace(self, connection): self._connecting.add(connection.features.shard_id) self._session.submit(self._open_connection_to_missing_shard, connection.features.shard_id) else: - connection = self._session.cluster.connection_factory(self.endpoint, - on_orphaned_stream_released=self.on_orphaned_stream_released) + new_connection = self._session.cluster.connection_factory( + self.endpoint, + on_orphaned_stream_released=self.on_orphaned_stream_released) try: - self._sync_keyspace(connection) + self._sync_keyspace(new_connection) except Exception: - failed_connection = connection + failed_connection = new_connection raise except Exception: log.warning("Failed reconnecting %s. Retrying." % (self.endpoint,)) + self._replacement_connections[shard_id] = connection self._session.submit(self._replace, connection) else: - self._is_replacing = False + self._replacement_connections.pop(shard_id, None) with self._stream_available_condition: self._stream_available_condition.notify() if failed_connection is not None: failed_connection.close() - def shutdown(self): + def shutdown(self, _failure=None): log.debug("Shutting down connections to %s", self.host) - with self._lock: + shutdown_lock = _failure.locked() if _failure is not None else self._lock + with shutdown_lock: + if (_failure is not None and not _failure.owns_locked()): + return with self._membership_lock: if self.is_shutdown: return + if _failure is not None: + # Only this reservation may retain failure authority after + # its own guarded retirement clears the shard slots. + _failure.retired = True self.is_shutdown = True connections_to_close = self._connections.copy() self._connections.clear() with self._stream_available_condition: self._stream_available_condition.notify_all() - for future in self._shard_connections_futures: - future.cancel() - + futures_to_cancel = tuple(self._shard_connections_futures) + self._replacement_connections.clear() pending_connections_to_close = self._pending_connections.copy() self._pending_connections.clear() + for future in futures_to_cancel: + future.cancel() + # connection.close can call pool.return_connection, which will # obtain self._lock via self._stream_available_condition. # So, it never should be called within self._lock context diff --git a/tests/unit/test_host_connection_pool.py b/tests/unit/test_host_connection_pool.py index b93e99bd14..34bd57737b 100644 --- a/tests/unit/test_host_connection_pool.py +++ b/tests/unit/test_host_connection_pool.py @@ -11,7 +11,7 @@ # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. -from concurrent.futures import ThreadPoolExecutor +from concurrent.futures import Future, ThreadPoolExecutor import logging import uuid from cassandra.protocol_features import ProtocolFeatures @@ -19,10 +19,11 @@ from cassandra.shard_info import _ShardingInfo import unittest -from threading import Thread, Event, Lock +from threading import Barrier, Thread, Event, Lock, RLock from unittest.mock import Mock, NonCallableMagicMock, MagicMock, patch -from cassandra.cluster import Session, ShardAwareOptions +from cassandra.cluster import Cluster, Session, ShardAwareOptions +from cassandra._lifecycle import ConnectionFailureReservation from cassandra.connection import Connection, ConnectionBusy, ConnectionException, ConnectionSendError, DefaultEndPoint from cassandra.pool import HostConnection from cassandra.pool import Host, NoConnectionsAvailable @@ -30,42 +31,18 @@ import pytest from tests.unit.util import HashableMock +from tests.unit.utils import new_session_with_pool_state LOGGER = logging.getLogger(__name__) -class _ExpectedStaleSocketFailure(Exception): - pass - - -@pytest.mark.parametrize('ownership, convicted, postcondition', [ - pytest.param('current', False, None), - pytest.param('current', True, None), -] + [ - pytest.param( - ownership, convicted, postcondition, - marks=pytest.mark.xfail( - strict=True, raises=_ExpectedStaleSocketFailure, - reason='#858: stale socket failures still affect current ownership' - ) if postcondition in expected_failures else ()) - for ownership, convicted, expected_failures in [ - ('draining', False, ('conviction', 'replacement_scheduled', - 'shard_connection')), - ('draining', True, ('conviction', 'down', 'pool_open', - 'shard_connection', 'replacement_open')), - ('retired_pool', False, ('conviction',)), - ('retired_pool', True, ('conviction', 'down')), - ] - for postcondition in [ - 'conviction', 'down', 'replacement_scheduled', 'pool_open', - 'shard_connection', 'replacement_open', - ] -]) -def test_connection_failure_respects_socket_ownership( - ownership, convicted, postcondition): +def _make_socket_pool(session=None): host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=uuid.uuid4()) - host.signal_connection_failure = Mock(return_value=convicted) - session = NonCallableMagicMock(spec=Session, keyspace=None) + host.signal_connection_failure = Mock(return_value=False) + if session is None: + session = NonCallableMagicMock( + spec=Session, keyspace=None, _lock=RLock(), is_shutdown=False) + session.cluster.is_shutdown = False session.cluster.shard_aware_options.disable = True def connection(): @@ -80,6 +57,25 @@ def connection(): session.cluster.connection_factory.side_effect = [original, replacement] pool = HostConnection(host, HostDistance.LOCAL, session) session._pools = {host: pool} + return host, session, pool, original, replacement + + +@pytest.fixture +def socket_pool(): + return _make_socket_pool() + + +@pytest.mark.parametrize('ownership', [ + 'current', 'draining', 'retired_pool', 'unpublished_pool', + 'retired_host', 'moved_endpoint', 'shutdown_session', +]) +@pytest.mark.parametrize('convicted', [False, True]) +@pytest.mark.parametrize('shutdown_on_error', [False, True]) +def test_connection_failure_respects_socket_ownership( + socket_pool, ownership, convicted, shutdown_on_error): + host, session, pool, original, replacement = socket_pool + host.signal_connection_failure.return_value = convicted + pool.shutdown_on_error = shutdown_on_error borrowed, _ = pool.borrow_connection(timeout=0) assert borrowed is original assert original.in_flight == 1 @@ -89,10 +85,17 @@ def connection(): # slot has been given to a healthy replacement. pool._connections[0] = replacement pool._trash.add(original) - elif ownership == 'retired_pool': + elif ownership in ('retired_pool', 'unpublished_pool'): winner = HostConnection(host, HostDistance.LOCAL, session) session._pools = {host: winner} - pool.shutdown() + if ownership == 'retired_pool': + pool.shutdown() + elif ownership == 'retired_host': + host._is_removed = True + elif ownership == 'moved_endpoint': + host.endpoint = DefaultEndPoint('127.0.0.2') + elif ownership == 'shutdown_session': + session.is_shutdown = True original.is_defunct = True original.last_error = ConnectionException('old socket failed', host.endpoint) @@ -103,29 +106,341 @@ def connection(): assert replacement.in_flight == 0 if ownership == 'current': host.signal_connection_failure.assert_called_once_with(original.last_error) - assert pool.is_shutdown is convicted - if convicted: - session.cluster.on_down.assert_called_once_with( - host, is_host_addition=False) + assert pool.is_shutdown is (convicted or shutdown_on_error) + if convicted or shutdown_on_error: + session.cluster.on_down.assert_called_once() + args, kwargs = session.cluster.on_down.call_args + assert args == (host,) + assert kwargs['is_host_addition'] is False else: session.submit.assert_called_once_with(pool._replace, original) else: current = session._pools[host] + host.signal_connection_failure.assert_not_called() + session.cluster.on_down.assert_not_called() + session.submit.assert_not_called() + assert not current.is_shutdown + if ownership in ('draining', 'retired_pool', 'unpublished_pool'): + assert current._connections.get(0) is replacement + replacement.close.assert_not_called() + + +@pytest.mark.parametrize('convicted', [False, True]) +@pytest.mark.parametrize('change', ['socket', 'pool', 'endpoint', 'retirement']) +def test_failure_policy_cannot_affect_replacement(socket_pool, convicted, change): + host, session, pool, original, replacement = socket_pool + pool.borrow_connection(timeout=0) + original.is_defunct = True + original.last_error = ConnectionException('socket failed', host.endpoint) + changed = Event() + errors = [] + + def replace(): try: - if postcondition == 'conviction': - host.signal_connection_failure.assert_not_called() - elif postcondition == 'down': - session.cluster.on_down.assert_not_called() - elif postcondition == 'replacement_scheduled': - session.submit.assert_not_called() - elif postcondition == 'pool_open': - assert not current.is_shutdown - elif postcondition == 'shard_connection': - assert current._connections.get(0) is replacement - elif postcondition == 'replacement_open': - replacement.close.assert_not_called() - except AssertionError as error: - raise _ExpectedStaleSocketFailure from error + winner = (HostConnection(host, HostDistance.LOCAL, session) + if change == 'pool' else None) + with original.lock, session._lock, host.lock, pool._lock: + if change == 'socket': + pool._connections[0] = replacement + pool._trash.add(original) + elif change == 'pool': + session._pools[host] = winner + elif change == 'endpoint': + host.endpoint = DefaultEndPoint('127.0.0.2') + else: + host._is_removed = True + except Exception as error: + errors.append(error) + finally: + changed.set() + + worker = Thread(target=replace) + + def convict(error): + worker.start() + assert changed.wait(5), 'policy ran while ownership locks were held' + assert not errors + return convicted + + host.signal_connection_failure.side_effect = convict + try: + pool.return_connection(original) + finally: + worker.join(5) + assert not worker.is_alive() + assert original.in_flight == 0 + session.cluster.on_down.assert_not_called() + session.submit.assert_not_called() + assert not session._pools[host].is_shutdown + if change in ('socket', 'pool'): + assert session._pools[host]._connections[0] is replacement + replacement.close.assert_not_called() + + +@pytest.mark.parametrize('change', ['pool', 'endpoint', 'health']) +def test_down_admission_rechecks_failed_socket(socket_pool, change): + host, session, pool, original, replacement = socket_pool + cluster = Cluster(contact_points=[]) + session.cluster = cluster + cluster.sessions = (session,) + cluster.profile_manager = Mock() + cluster.on_down_potentially_blocking = Mock(return_value=Future()) + host.set_up() + host.signal_connection_failure.return_value = True + pool.borrow_connection(timeout=0) + original.is_defunct = True + original.last_error = ConnectionException('socket failed', host.endpoint) + + def distance(current): + with session._lock, host.lock, pool._lock: + if change == 'pool': + session._pools[host] = Mock( + host=host, endpoint=host.endpoint, is_shutdown=False) + elif change == 'endpoint': + host.endpoint = DefaultEndPoint('127.0.0.2') + else: + host._health_generation += 1 + return HostDistance.IGNORED + + cluster.profile_manager.distance.side_effect = distance + try: + pool.return_connection(original) + assert host.is_up + assert pool.is_shutdown + if change == 'pool': + assert not session._pools[host].is_shutdown + replacement.close.assert_not_called() + cluster.on_down_potentially_blocking.assert_not_called() + finally: + pool.shutdown() + cluster.shutdown() + + +@pytest.mark.parametrize('change', ['pool', 'endpoint', 'host', 'session', 'cluster']) +def test_retirement_cleanup_cannot_down_superseding_owner(socket_pool, change): + host, session, pool, original, replacement = socket_pool + host.signal_connection_failure.return_value = True + pool.borrow_connection(timeout=0) + original.is_defunct = True + successor = Mock(is_shutdown=False) + + def supersede(): + # Socket cleanup is outside ownership guards, so it can synchronously + # change topology or publish a successor before DOWN inspection starts. + with session._lock, host.lock: + assert pool._lock.acquire(False), 'cleanup retained the pool work lock' + try: + if change == 'pool': + session._pools[host] = successor + elif change == 'endpoint': + host.endpoint = DefaultEndPoint('127.0.0.2') + elif change == 'host': + host._is_removed = True + elif change == 'session': + session.is_shutdown = True + else: + session.cluster.is_shutdown = True + finally: + pool._lock.release() + + original.close.side_effect = supersede + pool.return_connection(original) + assert pool.is_shutdown + assert original.in_flight == 0 + original.close.assert_called_once() + session.cluster.on_down.assert_not_called() + replacement.close.assert_not_called() + successor.shutdown.assert_not_called() + + +@pytest.mark.parametrize('retirement', ['owned', 'generic', 'already_shutdown', 'stale', 'other_failure']) +def test_only_own_guarded_retirement_can_admit_down(socket_pool, retirement): + host, session, pool, original, replacement = socket_pool + cluster = Cluster(contact_points=[]) + session.cluster = cluster + cluster.sessions = (session,) + cluster.profile_manager.distance = Mock(return_value=HostDistance.IGNORED) + cluster.on_down_potentially_blocking = Mock(return_value=Future()) + host.set_up() + failure = ConnectionFailureReservation(pool, original) + try: + if retirement == 'generic': + pool.shutdown() + elif retirement == 'already_shutdown': + pool.shutdown() + pool.shutdown(_failure=failure) + elif retirement == 'stale': + pool._connections[0] = replacement + pool.shutdown(_failure=failure) + elif retirement == 'other_failure': + pool.shutdown(_failure=ConnectionFailureReservation(pool, original)) + pool.shutdown(_failure=failure) + else: + pool.shutdown(_failure=failure) + + admitted = cluster.on_down(host, False, _pool_failure=failure) + assert admitted is (retirement == 'owned') + assert host.is_up is (retirement != 'owned') + assert cluster.on_down_potentially_blocking.call_count == int(retirement == 'owned') + finally: + pool.shutdown() + cluster.shutdown() + + +def test_racing_first_callbacks_preserve_convicting_owner(socket_pool): + host, session, pool, original, _ = socket_pool + original.in_flight = 2 + original.is_defunct = True + start_callbacks = Barrier(3) + policy_entered, release_policy, duplicate_done = Event(), Event(), Event() + errors = [] + + def convict(error): + policy_entered.set() + assert release_policy.wait(5) + return True + + def fail(): + try: + start_callbacks.wait(timeout=5) + pool.return_connection(original) + except BaseException as exc: + errors.append(exc) + finally: + duplicate_done.set() + + host.signal_connection_failure.side_effect = convict + workers = (Thread(target=fail), Thread(target=fail)) + try: + for worker in workers: + worker.start() + start_callbacks.wait(timeout=5) + assert policy_entered.wait(5) + assert duplicate_done.wait(2), 'duplicate callback waited for conviction' + assert pool._connections[0] is original + finally: + release_policy.set() + start_callbacks.abort() + for worker in workers: + if worker.ident is not None: + worker.join(5) + pool.shutdown() + assert not errors + assert all(not worker.is_alive() for worker in workers) + assert original.in_flight == 0 + host.signal_connection_failure.assert_called_once() + session.cluster.on_down.assert_called_once() + session.submit.assert_not_called() + + +@pytest.mark.parametrize('independent_pool', [False, True]) +def test_failed_pool_cannot_discount_its_own_down(independent_pool): + session = new_session_with_pool_state() + session.cluster = Mock() + session.keyspace = None + session.is_shutdown = False + host, session, pool, original, sibling = _make_socket_pool(session) + sibling.features = ProtocolFeatures(shard_id=1) + pool._connections[1] = sibling + + other_session = new_session_with_pool_state() + other_session.cluster = session.cluster + other_session.keyspace = None + other_session.is_shutdown = False + healthy = HashableMock( + spec=Connection, in_flight=0, is_closed=False, is_defunct=False, + orphaned_request_ids=set(), features=ProtocolFeatures(shard_id=0)) + session.cluster.connection_factory.side_effect = [healthy] + other_pool = HostConnection(host, HostDistance.LOCAL, other_session) + other_session._pools[host] = other_pool + + cluster = Cluster(contact_points=[]) + session.cluster = other_session.cluster = cluster + cluster.sessions = ((session, other_session) if independent_pool + else (session,)) + cluster.profile_manager.distance = Mock(return_value=HostDistance.LOCAL) + cluster.on_down_potentially_blocking = Mock(return_value=Future()) + host.set_up() + host.signal_connection_failure.return_value = True + original.in_flight = 1 + original.is_defunct = True + original.last_error = ConnectionException('one shard failed', host.endpoint) + try: + assert cluster._discount_down_events + assert session.get_pool_state()[host]['open_count'] == 1 + pool.return_connection(original) + assert original.in_flight == 0 + assert pool.is_shutdown + assert host.is_up is independent_pool + assert (cluster.on_down_potentially_blocking.call_count == + (0 if independent_pool else 1)) + assert not other_pool.is_shutdown + healthy.close.assert_not_called() + finally: + pool.shutdown() + other_pool.shutdown() + cluster.shutdown() + + +def test_queued_replacement_cannot_remove_new_socket(socket_pool): + host, session, pool, original, replacement = socket_pool + pool.borrow_connection(timeout=0) + original.is_defunct = True + pool.return_connection(original) + session.submit.assert_called_once_with(pool._replace, original) + + with pool._lock: + pool._connections[0] = replacement + pool._replace(original) + assert pool._connections[0] is replacement + assert session.cluster.connection_factory.call_count == 1 + replacement.close.assert_not_called() + + # The stale worker must neither suppress a new failure nor clear its owner. + pool.borrow_connection(timeout=0) + replacement.is_defunct = True + pool.return_connection(replacement) + session.submit.assert_called_with(pool._replace, replacement) + pool._replace(original) + healthy = HashableMock(spec=Connection) + healthy.features = ProtocolFeatures(shard_id=0) + session.cluster.connection_factory.side_effect = [healthy] + pool._replace(replacement) + assert pool._connections[0] is healthy + + +def test_current_pool_still_shuts_down_when_down_dispatch_raises(socket_pool): + host, session, pool, original, _ = socket_pool + host.signal_connection_failure.return_value = True + session.cluster.on_down.side_effect = RuntimeError('executor stopped') + pool.borrow_connection(timeout=0) + original.is_defunct = True + + with pytest.raises(RuntimeError, match='executor stopped'): + pool.return_connection(original) + + assert pool.is_shutdown + assert original.in_flight == 0 + original.close.assert_called_once() + + +def test_return_after_session_collection_releases_original_stream(): + import gc + import weakref + + host, session, pool, original, _ = _make_socket_pool() + pool.borrow_connection(timeout=0) + original.is_defunct = True + session_ref = weakref.ref(session) + del session + gc.collect() + assert session_ref() is None + + pool.return_connection(original) + + assert original.in_flight == 0 + original.close.assert_called_once() + host.signal_connection_failure.assert_not_called() class _PoolTests(unittest.TestCase): @@ -134,16 +449,26 @@ class _PoolTests(unittest.TestCase): uses_single_connection = None def make_session(self): - session = NonCallableMagicMock(spec=Session, keyspace='foobarkeyspace', _trash=[]) + session = NonCallableMagicMock( + spec=Session, keyspace='foobarkeyspace', _trash=[], + _lock=RLock(), _pools={}, is_shutdown=False) + session.cluster.is_shutdown = False return session + def make_pool(self, host, session): + host.lock = RLock() + host._is_removed = False + pool = self.PoolImpl(host, HostDistance.LOCAL, session) + session._pools[host] = pool + return pool + def test_borrow_and_return(self): host = Mock(spec=Host, address='ip1') session = self.make_session() conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100) session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released) c, request_id = pool.borrow_connection(timeout=0.01) @@ -165,7 +490,7 @@ def test_init_logs_none_shard_id(self): session.cluster.connection_factory.return_value = conn with self.assertLogs('cassandra.pool', level='DEBUG') as logs: - self.PoolImpl(host, HostDistance.LOCAL, session) + self.make_pool(host, session) shard_logs = [r.getMessage() for r in logs.records if "First connection created" in r.getMessage()] @@ -178,7 +503,7 @@ def test_failed_wait_for_connection(self): conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100) session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released) pool.borrow_connection(timeout=0.01) @@ -198,7 +523,7 @@ def test_successful_wait_for_connection(self): lock=Lock()) session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released) pool.borrow_connection(timeout=0.01) @@ -223,7 +548,7 @@ def test_spawn_when_at_max(self): conn.max_request_id = 100 session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released) pool.borrow_connection(timeout=0.01) @@ -247,7 +572,7 @@ def test_return_defunct_connection(self): max_request_id=100, signaled_error=False) session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released) pool.borrow_connection(timeout=0.01) @@ -267,7 +592,7 @@ def test_replace_only_does_not_mark_host_down(self): is_closed=False, max_request_id=100, signaled_error=False) session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) pool.borrow_connection(timeout=0.01) conn.is_defunct = True @@ -286,7 +611,7 @@ def test_signaled_error_still_shuts_down_when_configured(self): is_closed=False, max_request_id=100, signaled_error=False) session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) pool.shutdown_on_error = True pool.borrow_connection(timeout=0.01) @@ -304,7 +629,7 @@ def test_local_send_error_skips_shutdown_when_configured(self): is_closed=False, max_request_id=100, signaled_error=False) session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) pool.shutdown_on_error = True pool.borrow_connection(timeout=0.01) @@ -326,7 +651,7 @@ def test_heartbeat_failure_shuts_down_after_local_send_error(self): is_closed=False, max_request_id=100, signaled_error=False) session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) pool.shutdown_on_error = True pool.borrow_connection(timeout=0.01) @@ -337,7 +662,8 @@ def test_heartbeat_failure_shuts_down_after_local_send_error(self): assert pool.is_shutdown host.signal_connection_failure.assert_not_called() - session.cluster.on_down.assert_called_once_with(host, is_host_addition=False) + session.cluster.on_down.assert_called_once_with( + host, is_host_addition=False, _pool_failure=unittest.mock.ANY) def test_return_defunct_connection_on_down_host(self): host = Mock(spec=Host, address='ip1') @@ -348,7 +674,7 @@ def test_return_defunct_connection_on_down_host(self): session.cluster.connection_factory.return_value = conn session.cluster.shard_aware_options = ShardAwareOptions() - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released) pool.borrow_connection(timeout=0.01) @@ -375,7 +701,7 @@ def test_return_closed_connection(self): signaled_error=False, orphaned_threshold_reached=False) session.cluster.connection_factory.return_value = conn - pool = self.PoolImpl(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) session.cluster.connection_factory.assert_called_once_with(host.endpoint, on_orphaned_stream_released=pool.on_orphaned_stream_released) pool.borrow_connection(timeout=0.01) @@ -471,7 +797,7 @@ class HostConnectionTests(_PoolTests): PoolImpl = HostConnection uses_single_connection = True - def test_replacement_connection_uses_pool_endpoint(self): + def test_replacement_skips_pool_after_endpoint_changes(self): host = Host('127.0.0.1', SimpleConvictionPolicy, host_id=uuid.uuid4()) session = self.make_session() session.keyspace = None @@ -483,14 +809,16 @@ def test_replacement_connection_uses_pool_endpoint(self): replacement = HashableMock(spec=Connection) replacement.features.shard_id = 0 session.cluster.connection_factory.side_effect = [first, replacement] - pool = HostConnection(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) original_endpoint = host.endpoint host.endpoint = DefaultEndPoint('127.0.0.2') pool._replace(first) assert pool.endpoint == original_endpoint - assert session.cluster.connection_factory.call_args.args[0] == original_endpoint + assert session.cluster.connection_factory.call_count == 1 + assert pool._connections[0] is first + replacement.close.assert_not_called() pool.shutdown() def test_borrow_rejects_unsynchronized_connection_without_waiting(self): @@ -499,7 +827,7 @@ def test_borrow_rejects_unsynchronized_connection_without_waiting(self): conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100, lock=Lock()) session.cluster.connection_factory.return_value = conn - pool = HostConnection(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) pool._keyspace = 'newks' conn.keyspace = 'foobarkeyspace' conn.set_keyspace_blocking.side_effect = \ @@ -520,7 +848,7 @@ def test_failed_keyspace_change_can_be_corrected_by_use(self): conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100, lock=Lock()) session.cluster.connection_factory.return_value = conn - pool = HostConnection(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) errors = [] pool._set_keyspace_for_all_conns( @@ -544,7 +872,7 @@ def test_failed_keyspace_send_still_completes_pool_callback(self): conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100, lock=Lock()) session.cluster.connection_factory.return_value = conn - pool = HostConnection(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) error = ConnectionBusy('send blocked') conn.set_keyspace_async.side_effect = error callback = Mock() @@ -562,7 +890,7 @@ def test_stale_shard_uses_aligned_connection(self): is_closed=False, max_request_id=100, lock=Lock()) stale.features.shard_id = 0 session.cluster.connection_factory.return_value = stale - pool = HostConnection(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) aligned = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100, lock=Lock()) aligned.features.shard_id = 1 @@ -581,7 +909,7 @@ def test_published_connection_catches_newer_keyspace(self): conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100, lock=Lock()) session.cluster.connection_factory.return_value = conn - pool = HostConnection(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) conn.keyspace = None def set_keyspace(keyspace): @@ -601,7 +929,7 @@ def test_async_keyspace_reply_reapplies_current_target(self): conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, is_closed=False, max_request_id=100, lock=Lock()) session.cluster.connection_factory.return_value = conn - pool = HostConnection(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) finished = [] pool._set_keyspace_for_all_conns( @@ -624,18 +952,17 @@ def test_async_keyspace_reply_reapplies_current_target(self): assert conn.keyspace == pool._keyspace def test_failed_replacement_keyspace_sync_removes_connection(self): - host = Mock(spec=Host, address='ip1', sharding_info=None) - session = self.make_session() - session.cluster.shard_aware_options.disable = True - old_conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, - is_closed=False, max_request_id=100, lock=Lock()) - old_conn.features.shard_id = 0 - old_conn.features.sharding_info = None - new_conn = HashableMock(spec=Connection, in_flight=0, is_defunct=False, - is_closed=False, max_request_id=100, lock=Lock()) - new_conn.features.shard_id = 0 - session.cluster.connection_factory.side_effect = [old_conn, new_conn] - pool = HostConnection(host, HostDistance.LOCAL, session) + host, session, pool, old_conn, new_conn = _make_socket_pool() + pool._keyspace = 'foobarkeyspace' + # Regular-port replacements can land on a different shard. + new_conn.features = ProtocolFeatures(shard_id=1) + healthy = HashableMock( + spec=Connection, in_flight=0, is_defunct=False, is_closed=False, + max_request_id=100, lock=Lock(), keyspace=None, + features=ProtocolFeatures(shard_id=0)) + healthy.set_keyspace_blocking.side_effect = \ + lambda keyspace: setattr(healthy, 'keyspace', keyspace) + session.cluster.connection_factory.side_effect = [new_conn, healthy] def set_keyspace(keyspace): if keyspace == 'foobarkeyspace': @@ -645,11 +972,26 @@ def set_keyspace(keyspace): raise ConnectionException('sync failed', host.endpoint) new_conn.set_keyspace_blocking.side_effect = set_keyspace - pool._replace(old_conn) - - assert 0 not in pool._connections - new_conn.close.assert_called_once() - session.submit.assert_called_once_with(pool._replace, new_conn) + old_conn.is_defunct = True + old_conn.in_flight = 1 + try: + pool.return_connection(old_conn) + replace, retired = session.submit.call_args.args + replace(retired) + + assert not pool._connections + new_conn.close.assert_called_once() + replace, retired = session.submit.call_args.args + replace(retired) + + borrowed, _ = pool.borrow_connection(timeout=0) + assert borrowed is healthy + assert healthy.keyspace == pool._keyspace == 'newks' + assert not pool._replacement_connections + healthy.close.assert_not_called() + pool.return_connection(borrowed) + finally: + pool.shutdown() def test_failed_shard_keyspace_sync_allows_retry(self): host = Mock(spec=Host, address='ip1', sharding_info=None) @@ -662,7 +1004,7 @@ def test_failed_shard_keyspace_sync_allows_retry(self): is_closed=False, max_request_id=100, lock=Lock()) shard_conn.features.shard_id = 1 session.cluster.connection_factory.side_effect = [first_conn, shard_conn] - pool = HostConnection(host, HostDistance.LOCAL, session) + pool = self.make_pool(host, session) host.sharding_info = Mock(shards_count=2) pool._get_shard_aware_endpoint = Mock(return_value=None) pool._connecting.add(1) diff --git a/tests/unit/test_lifecycle_coordination.py b/tests/unit/test_lifecycle_coordination.py index 0fee02480a..fe1927f408 100644 --- a/tests/unit/test_lifecycle_coordination.py +++ b/tests/unit/test_lifecycle_coordination.py @@ -13,13 +13,14 @@ # limitations under the License. from concurrent.futures import Future -from threading import Event, Lock, Thread +from threading import Barrier, Event, Lock, Thread, get_ident from unittest.mock import Mock, patch import pytest from cassandra.cluster import Cluster, ControlConnection -from cassandra._lifecycle import LifecycleCoordinator, PoolFailureReservation +from cassandra._lifecycle import ( + ConnectionFailureReservation, LifecycleCoordinator, PoolFailureReservation) from cassandra.connection import ( Connection, ConnectionException, DefaultEndPoint) from cassandra.policies import HostDistance @@ -964,6 +965,316 @@ def close(self): self.is_closed = True +@pytest.mark.parametrize('first_failure_waiting', [False, True], + ids=['already_claimed', 'first_waiting']) +def test_duplicate_failed_callback_allows_same_pool_use_response( + topology, first_failure_waiting): + from cassandra.protocol import ResultMessage, RESULT_KIND_SET_KEYSPACE + from cassandra.shard_info import _ShardingInfo + + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + session = new_session_with_pool_state() + session.cluster = cluster + session.keyspace = None + session.is_shutdown = False + session.submit = Mock() + original = _PublicationConnection(host.endpoint, compression=False) + original.features = ProtocolFeatures(shard_id=1 if first_failure_waiting else 0) + candidate = _PublicationConnection(host.endpoint, compression=False) + candidate.features = ProtocolFeatures(shard_id=0) + cluster.connection_factory = Mock(side_effect=[original, candidate]) + cluster.shard_aware_options.disable_shardaware_port = True + pool = HostConnection(host, HostDistance.LOCAL, session) + session._pools[host] = pool + session.keyspace = pool._keyspace = original.keyspace = 'ks' + host.sharding_info = _ShardingInfo(0, 2, '', '', 0, None, None) + host.signal_connection_failure = Mock(return_value=False) + original.in_flight = 2 + original.last_error = ConnectionResetError('socket reset') + if not first_failure_waiting: + original.is_defunct = True + pool.return_connection(original) + assert original.in_flight == 1 + session.submit.assert_called_once_with(pool._replace, original) + pool._replace(original) + assert session.submit.call_count == 2 + session.submit.assert_called_with(pool._open_connection_to_missing_shard, 0) + + io_waiting = Event() + first_blocking = Event() + reactor_done = Event() + response_lock = Lock() + responses = [] + cleanup_started = False + errors = [] + + if first_failure_waiting: + work_lock = pool._lock + + class WorkLock: + def acquire(self, blocking=True): + # A nonblocking ownership probe is not the common checkpoint: + # the first callback must actually wait behind pending USE. + if blocking and get_ident() == first.ident: + first_blocking.set() + return work_lock.acquire(blocking) + + def release(self): + work_lock.release() + + def __enter__(self): + self.acquire() + return self + + def __exit__(self, *args): + self.release() + + pool._lock = WorkLock() + + def send(message, request_id, cb, **kwargs): + assert message.query.startswith('USE ') + with response_lock: + respond_now = cleanup_started + if not respond_now: + responses.append(cb) + if respond_now: + cb(ResultMessage(RESULT_KIND_SET_KEYSPACE)) + io_waiting.set() + return 0 + + def respond(cleanup=False): + nonlocal cleanup_started + with response_lock: + cleanup_started |= cleanup + callbacks = list(responses) + responses.clear() + for callback in callbacks: + callback(ResultMessage(RESULT_KIND_SET_KEYSPACE)) + + def run(fn): + try: + fn() + except BaseException as exc: + errors.append(exc) + + def process_responses(): + # The same reactor must finish this redundant failure callback before + # delivering the USE response that releases the replacement work lock. + pool.return_connection(original) + respond() + reactor_done.set() + + candidate.send_msg = send + worker = Thread(target=run, args=(lambda: pool._open_connection_to_missing_shard(0),)) + first = Thread(target=run, args=(lambda: pool.return_connection(original),)) + reactor = Thread(target=run, args=(process_responses,)) + threads = (worker, first, reactor) if first_failure_waiting else (worker, reactor) + try: + try: + worker.start() + assert io_waiting.wait(5) + if first_failure_waiting: + original.is_defunct = True + first.start() + assert first_blocking.wait(5), 'first failure did not wait for USE' + reactor.start() + progressed = reactor_done.wait(2) + finally: + respond(cleanup=True) + for thread in threads: + if thread.ident is not None: + thread.join(5) + assert all(not thread.is_alive() for thread in threads) + assert not errors + assert original.in_flight == candidate.in_flight == 0 + assert candidate.keyspace == 'ks' + host.signal_connection_failure.assert_called_once_with(original.last_error) + assert host.is_up + assert not pool.is_shutdown + if first_failure_waiting: + session.submit.assert_called_once_with(pool._replace, original) + assert session._pools[host] is pool + borrowed, _ = pool.borrow_connection(timeout=0) + assert borrowed is candidate + pool.return_connection(borrowed) + assert candidate.in_flight == 0 + else: + assert session.submit.call_count == 2 + assert progressed, 'duplicate failure blocked same-pool USE response' + finally: + pool.shutdown() + + +@pytest.mark.parametrize('independent_healthy_pool', [False, True]) +def test_concurrent_session_failures_retire_before_health_inspection( + topology, independent_healthy_pool): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + sessions, pools, failed, siblings = [], [], [], [] + for index in range(2 + int(independent_healthy_pool)): + session = new_session_with_pool_state() + session.cluster = cluster + session.keyspace = None + session.is_shutdown = False + session.shutdown = Mock() + session.on_down = Mock() + connection = _PublicationConnection(host.endpoint, compression=False) + connection.features = ProtocolFeatures(shard_id=0) + cluster.connection_factory = Mock(return_value=connection) + pool = HostConnection(host, HostDistance.LOCAL, session) + session._pools[host] = pool + sessions.append(session) + pools.append(pool) + if index < 2: + sibling = _PublicationConnection(host.endpoint, compression=False) + sibling.features = ProtocolFeatures(shard_id=1) + pool._connections[1] = sibling + siblings.append(sibling) + connection.is_defunct = True + connection.in_flight = 1 + connection.last_error = ConnectionException('shard failed', host.endpoint) + failed.append(connection) + cluster.sessions = tuple(sessions) + host.signal_connection_failure = Mock(return_value=True) + cluster.control_connection.on_down = Mock() + snapshots = Barrier(2) + for session in sessions[:2]: + get_state = session.get_pool_state + + def inspect(get_state=get_state): + state = get_state() + # Both failures capture the other session's real pool health + # before either can finish DOWN admission or its old cleanup. + snapshots.wait(timeout=5) + return state + + session.get_pool_state = inspect + + errors, scheduled, order = [], [], [] + cluster.profile_manager.on_down.side_effect = lambda _: order.append('down') + cluster._start_reconnector.side_effect = lambda *a, **kw: order.append('recovery') + + def delayed_submit(fn, *args, **kwargs): + scheduled.append((fn, args, kwargs)) + return Future() + + def fail(index): + try: + pools[index].return_connection(failed[index]) + except BaseException as exc: + errors.append(exc) + + workers = [Thread(target=fail, args=(index,)) for index in range(2)] + try: + with patch.object(cluster.executor, 'submit', side_effect=delayed_submit): + for worker in workers: + worker.start() + for worker in workers: + worker.join(7) + assert all(not worker.is_alive() for worker in workers) + assert not errors + assert all(pool.is_shutdown for pool in pools[:2]) + assert all(connection.is_closed for connection in failed + siblings) + assert all(connection.in_flight == 0 for connection in failed) + assert host.is_up is independent_healthy_pool + assert len(scheduled) == int(not independent_healthy_pool) + # Check actual FIFO delivery and recovery, including a later + # notification. A stale second failure must not duplicate DOWN. + with cluster._lifecycle.transaction(drain=False): + cluster._lifecycle.submit(order.append, 'later') + cluster._lifecycle.drain() + for fn, args, kwargs in scheduled: + fn(*args, **kwargs) + assert order == (['later'] if independent_healthy_pool else + ['down', 'recovery', 'later']) + assert cluster._start_reconnector.call_count == int(not independent_healthy_pool) + if independent_healthy_pool: + healthy, _ = pools[2].borrow_connection(timeout=0) + assert not healthy.is_closed + pools[2].return_connection(healthy) + assert not pools[2].is_shutdown + finally: + snapshots.abort() + for worker in workers: + if worker.ident is not None: + worker.join(5) + for pool in pools: + pool.shutdown() + + +@pytest.mark.parametrize('supersede', [ + None, 'pool', 'endpoint', 'host', 'session', 'cluster', 'health', +]) +def test_retired_failure_admission_does_not_wait_for_pool_work( + topology, supersede): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + session = new_session_with_pool_state() + session.cluster = cluster + session.keyspace = None + session.is_shutdown = False + session.shutdown = Mock() + connection = _PublicationConnection(host.endpoint, compression=False) + connection.features = ProtocolFeatures(shard_id=0) + cluster.connection_factory = Mock(return_value=connection) + pool = HostConnection(host, HostDistance.LOCAL, session) + session._pools[host] = pool + cluster.sessions = (session,) + successor = HostConnection(host, HostDistance.IGNORED, session) + reservation = ConnectionFailureReservation(pool, connection) + pool.shutdown(_failure=reservation) + cluster.on_down_potentially_blocking = Mock(return_value=Future()) + + def distance(_host): + with session._lock, host.lock: + if supersede == 'pool': + session._pools[host] = successor + elif supersede == 'endpoint': + host.endpoint = DefaultEndPoint('127.0.0.2') + elif supersede == 'host': + host._is_removed = True + elif supersede == 'session': + session.is_shutdown = True + elif supersede == 'cluster': + cluster.is_shutdown = True + elif supersede == 'health': + assert cluster._lifecycle.mark_ready(host) + return HostDistance.LOCAL + + cluster.profile_manager.distance.side_effect = distance + done = Event() + admitted, errors = [], [] + + def admit(): + try: + admitted.append(cluster.on_down(host, False, _pool_failure=reservation)) + except BaseException as exc: + errors.append(exc) + finally: + done.set() + + worker = Thread(target=admit) + try: + # After retirement the shard map has no authority left to inspect. + # Its work lock must not delay admission or hide superseding topology. + with pool._lock: + worker.start() + assert done.wait(2), 'retired failure waited for pool work' + worker.join(5) + assert not worker.is_alive() + assert not errors + assert admitted == [supersede is None] + assert host.is_up is (supersede is not None) + assert cluster.on_down_potentially_blocking.call_count == int(supersede is None) + assert not successor.is_shutdown + finally: + worker.join(5) + cluster.is_shutdown = False + pool.shutdown() + successor.shutdown() + + @pytest.mark.parametrize('failure', ['close', 'defunct']) def test_failed_candidate_cannot_fence_down_before_publication( topology, failure): @@ -1122,6 +1433,219 @@ def reply(): pool.shutdown() +def test_unpublished_failure_keeps_session_available_for_publication(topology): + cluster, _ = topology + host = cluster.metadata.get_host_by_host_id(HOST_ID_3) + session = new_session_with_pool_state() + session.cluster = cluster + session._profile_manager = cluster.profile_manager + session.keyspace = None + session.is_shutdown = False + session.shutdown = Mock() + connection = _PublicationConnection(host.endpoint, compression=False) + cluster.connection_factory = Mock(return_value=connection) + pool = HostConnection(host, HostDistance.LOCAL, session) + connection.is_defunct = True + host.signal_connection_failure = Mock() + work_lock = pool._lock + attempting = Event() + published = Event() + errors = [] + + class ObservedLock: + def __enter__(self): + attempting.set() + work_lock.acquire() + + def __exit__(self, *args): + work_lock.release() + + pool._lock = ObservedLock() + + def fail(): + try: + pool.return_connection(connection, stream_was_orphaned=True) + except BaseException as exc: + errors.append(exc) + + def publish(): + try: + with patch('cassandra.cluster.HostConnection', return_value=pool): + assert session.add_or_renew_pool(host, False).result(timeout=5) + published.set() + except BaseException as exc: + errors.append(exc) + + failure = Thread(target=fail) + publication = Thread(target=publish) + try: + # A background shard may own the work lock while publication takes + # membership then Session. A stale failure must release Session before + # waiting for work-lock cleanup, or all three operations can deadlock. + work_lock.acquire() + try: + failure.start() + assert attempting.wait(5) + publication.start() + assert published.wait(2) + finally: + work_lock.release() + finally: + for worker in (failure, publication): + if worker.ident is not None: + worker.join(5) + pool.shutdown() + assert not failure.is_alive() + assert not publication.is_alive() + assert not errors + assert session._pools[host] is pool + host.signal_connection_failure.assert_not_called() + + +@pytest.mark.parametrize('supersede', [None, 'pool']) +def test_current_failure_contention_allows_other_host_response( + topology, supersede): + from cassandra.protocol import ResultMessage, RESULT_KIND_SET_KEYSPACE + from cassandra.shard_info import _ShardingInfo + + cluster, _ = topology + hosts = [cluster.metadata.get_host_by_host_id(host_id) + for host_id in (HOST_ID_1, HOST_ID_2)] + session = new_session_with_pool_state() + session.cluster = cluster + session._profile_manager = cluster.profile_manager + session.keyspace = 'ks' + session.is_shutdown = False + session.shutdown = Mock() + cluster.shard_aware_options.disable_shardaware_port = True + connections = [_PublicationConnection(host.endpoint, compression=False) + for host in hosts] + for connection in connections: + connection.features = ProtocolFeatures(shard_id=0) + connection.keyspace = 'ks' + connection.is_defunct = True + connection.in_flight = 1 + candidate = _PublicationConnection(hosts[0].endpoint, compression=False) + candidate.features = ProtocolFeatures(shard_id=1) + cluster.connection_factory = Mock(side_effect=connections + [candidate]) + pools = [HostConnection(host, HostDistance.LOCAL, session) for host in hosts] + session._pools.update(zip(hosts, pools)) + hosts[0].sharding_info = _ShardingInfo(0, 2, '', '', 0, None, None) + successor = (HostConnection(hosts[0], HostDistance.IGNORED, session) + if supersede == 'pool' else None) + for host in hosts: + host.signal_connection_failure = Mock(return_value=True) + + io_waiting = Event() + contended = Event() + reactor_done = Event() + response_lock = Lock() + responses = [] + cleanup_started = False + delivered = [] + delivery_futures = [] + errors = [] + + def send(message, request_id, cb, **kwargs): + with response_lock: + respond_now = cleanup_started + if not respond_now: + responses.append(cb) + if respond_now: + cb(ResultMessage(RESULT_KIND_SET_KEYSPACE)) + io_waiting.set() + return 0 + + candidate.send_msg = send + + def respond(cleanup=False): + nonlocal cleanup_started + with response_lock: + cleanup_started |= cleanup + callbacks = list(responses) + responses.clear() + for callback in callbacks: + callback(ResultMessage(RESULT_KIND_SET_KEYSPACE)) + + def admit(host, is_addition, generation): + future = cluster._lifecycle.submit_async( + cluster.executor, delivered.append, host) + delivery_futures.append(future) + return future + + cluster.on_down_potentially_blocking = Mock(side_effect=admit) + work_lock = pools[0]._lock + + class ObservedLock: + def acquire(self, blocking=True): + if failure.ident == get_ident() and io_waiting.is_set(): + contended.set() + return work_lock.acquire(blocking) + + def release(self): + work_lock.release() + + def __enter__(self): + self.acquire() + + def __exit__(self, *args): + self.release() + + pools[0]._lock = ObservedLock() + + def run(fn): + try: + fn() + except BaseException as exc: + errors.append(exc) + + def process_responses(): + # One reactor must finish the other host's failure callback before + # it can deliver the USE response that releases the pool work lock. + if supersede == 'pool': + with session._lock, hosts[0].lock: + session._pools[hosts[0]] = successor + pools[1].return_connection(connections[1]) + respond() + reactor_done.set() + + failure = Thread(target=run, args=(lambda: pools[0].return_connection( + connections[0]),)) + worker = Thread(target=run, args=(lambda: pools[0]._open_connection_to_missing_shard(1),)) + reactor = Thread(target=run, args=(process_responses,)) + try: + worker.start() + assert io_waiting.wait(5) + failure.start() + assert contended.wait(5) + reactor.start() + assert reactor_done.wait(2), 'cross-host failure blocked reactor progress' + finally: + # Release controlled I/O even for the unfixed production negative + # control, so assertion failures cannot leave blocked driver threads. + respond(cleanup=True) + for thread in (worker, failure, reactor): + if thread.ident is not None: + thread.join(5) + for pool in pools: + pool.shutdown() + assert all(not thread.is_alive() for thread in (worker, failure, reactor)) + assert not errors + assert all(connection.in_flight == 0 for connection in connections) + for future in delivery_futures: + future.result(timeout=5) + assert delivered == ([hosts[1]] if supersede else [hosts[1], hosts[0]]) + assert hosts[0].is_up is bool(supersede) + assert not hosts[1].is_up + hosts[1].signal_connection_failure.assert_called_once() + assert hosts[0].signal_connection_failure.call_count == int(supersede != 'pool') + if successor is not None: + assert session._pools[hosts[0]] is successor + assert not successor.is_shutdown + successor.shutdown() + assert candidate.keyspace == 'ks' + + @pytest.mark.parametrize('healthy', [False, True]) def test_publication_progress_with_unsynchronized_background_shard( topology, healthy):