connection.py

# Copyright DataStax, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# 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 __future__ import absolute_import  # to enable import io from stdlib
from collections import defaultdict, deque
import errno
from functools import wraps, partial, total_ordering
from heapq import heappush, heappop
import io
import logging
import six
from six.moves import range
import socket
import struct
import sys
from threading import Thread, Event, RLock, Condition
import time
import ssl
import weakref


if 'gevent.monkey' in sys.modules:
    from gevent.queue import Queue, Empty
else:
    from six.moves.queue import Queue, Empty  # noqa

from cassandra import ConsistencyLevel, AuthenticationFailed, OperationTimedOut, ProtocolVersion
from cassandra.marshal import int32_pack
from cassandra.protocol import (ReadyMessage, AuthenticateMessage, OptionsMessage,
                                StartupMessage, ErrorMessage, CredentialsMessage,
                                QueryMessage, ResultMessage, ProtocolHandler,
                                InvalidRequestException, SupportedMessage,
                                AuthResponseMessage, AuthChallengeMessage,
                                AuthSuccessMessage, ProtocolException,
                                RegisterMessage, ReviseRequestMessage)
from cassandra.segment import SegmentCodec, CrcException
from cassandra.util import OrderedDict


log = logging.getLogger(__name__)

segment_codec_no_compression = SegmentCodec()
segment_codec_lz4 = None

# We use an ordered dictionary and specifically add lz4 before
# snappy so that lz4 will be preferred. Changing the order of this
# will change the compression preferences for the driver.
locally_supported_compressions = OrderedDict()

try:
    import lz4
except ImportError:
    pass
else:
    # The compress and decompress functions we need were moved from the lz4 to
    # the lz4.block namespace, so we try both here.
    try:
        from lz4 import block as lz4_block
    except ImportError:
        lz4_block = lz4

    try:
        lz4_block.compress
        lz4_block.decompress
    except AttributeError:
        raise ImportError(
            'lz4 not imported correctly. Imported object should have '
            '.compress and and .decompress attributes but does not. '
            'Please file a bug report on JIRA. (Imported object was '
            '{lz4_block})'.format(lz4_block=repr(lz4_block))
        )

    # Cassandra writes the uncompressed message length in big endian order,
    # but the lz4 lib requires little endian order, so we wrap these
    # functions to handle that

    def lz4_compress(byts):
        # write length in big-endian instead of little-endian
        return int32_pack(len(byts)) + lz4_block.compress(byts)[4:]

    def lz4_decompress(byts):
        # flip from big-endian to little-endian
        return lz4_block.decompress(byts[3::-1] + byts[4:])

    locally_supported_compressions['lz4'] = (lz4_compress, lz4_decompress)
    segment_codec_lz4 = SegmentCodec(lz4_compress, lz4_decompress)

try:
    import snappy
except ImportError:
    pass
else:
    # work around apparently buggy snappy decompress
    def decompress(byts):
        if byts == '\x00':
            return ''
        return snappy.decompress(byts)
    locally_supported_compressions['snappy'] = (snappy.compress, decompress)

DRIVER_NAME, DRIVER_VERSION = 'DataStax Python Driver', sys.modules['cassandra'].__version__

PROTOCOL_VERSION_MASK = 0x7f

HEADER_DIRECTION_FROM_CLIENT = 0x00
HEADER_DIRECTION_TO_CLIENT = 0x80
HEADER_DIRECTION_MASK = 0x80

frame_header_v1_v2 = struct.Struct('>BbBi')
frame_header_v3 = struct.Struct('>BhBi')


class EndPoint(object):
    """
    Represents the information to connect to a cassandra node.
    """

    @property
    def address(self):
        """
        The IP address of the node. This is the RPC address the driver uses when connecting to the node
        """
        raise NotImplementedError()

    @property
    def port(self):
        """
        The port of the node.
        """
        raise NotImplementedError()

    @property
    def ssl_options(self):
        """
        SSL options specific to this endpoint.
        """
        return None

    @property
    def socket_family(self):
        """
        The socket family of the endpoint.
        """
        return socket.AF_UNSPEC

    def resolve(self):
        """
        Resolve the endpoint to an address/port. This is called
        only on socket connection.
        """
        raise NotImplementedError()


class EndPointFactory(object):

    cluster = None

    def configure(self, cluster):
        """
        This is called by the cluster during its initialization.
        """
        self.cluster = cluster
        return self

    def create(self, row):
        """
        Create an EndPoint from a system.peers row.
        """
        raise NotImplementedError()


@total_ordering
class DefaultEndPoint(EndPoint):
    """
    Default EndPoint implementation, basically just an address and port.
    """

    def __init__(self, address, port=9042):
        self._address = address
        self._port = port

    @property
    def address(self):
        return self._address

    @property
    def port(self):
        return self._port

    def resolve(self):
        return self._address, self._port

    def __eq__(self, other):
        return isinstance(other, DefaultEndPoint) and \
               self.address == other.address and self.port == other.port

    def __hash__(self):
        return hash((self.address, self.port))

    def __lt__(self, other):
        return (self.address, self.port) < (other.address, other.port)

    def __str__(self):
        return str("%s:%d" % (self.address, self.port))

    def __repr__(self):
        return "<%s: %s:%d>" % (self.__class__.__name__, self.address, self.port)


class DefaultEndPointFactory(EndPointFactory):

    port = None
    """
    If no port is discovered in the row, this is the default port
    used for endpoint creation. 
    """

    def __init__(self, port=None):
        self.port = port

    def create(self, row):
        # TODO next major... move this class so we don't need this kind of hack
        from cassandra.metadata import _NodeInfo
        addr = _NodeInfo.get_broadcast_rpc_address(row)
        port = _NodeInfo.get_broadcast_rpc_port(row)
        if port is None:
            port = self.port if self.port else 9042

        # create the endpoint with the translated address
        # TODO next major, create a TranslatedEndPoint type
        return DefaultEndPoint(
            self.cluster.address_translator.translate(addr),
            port)


@total_ordering
class SniEndPoint(EndPoint):
    """SNI Proxy EndPoint implementation."""

    def __init__(self, proxy_address, server_name, port=9042):
        self._proxy_address = proxy_address
        self._index = 0
        self._resolved_address = None  # resolved address
        self._port = port
        self._server_name = server_name
        self._ssl_options = {'server_hostname': server_name}

    @property
    def address(self):
        return self._proxy_address

    @property
    def port(self):
        return self._port

    @property
    def ssl_options(self):
        return self._ssl_options

    def resolve(self):
        try:
            resolved_addresses = socket.getaddrinfo(self._proxy_address, self._port,
                                                    socket.AF_UNSPEC, socket.SOCK_STREAM)
        except socket.gaierror:
            log.debug('Could not resolve sni proxy hostname "%s" '
                      'with port %d' % (self._proxy_address, self._port))
            raise

        # round-robin pick
        self._resolved_address = sorted(addr[4][0] for addr in resolved_addresses)[self._index % len(resolved_addresses)]
        self._index += 1

        return self._resolved_address, self._port

    def __eq__(self, other):
        return (isinstance(other, SniEndPoint) and
                self.address == other.address and self.port == other.port and
                self._server_name == other._server_name)

    def __hash__(self):
        return hash((self.address, self.port, self._server_name))

    def __lt__(self, other):
        return ((self.address, self.port, self._server_name) <
                (other.address, other.port, self._server_name))

    def __str__(self):
        return str("%s:%d:%s" % (self.address, self.port, self._server_name))

    def __repr__(self):
        return "<%s: %s:%d:%s>" % (self.__class__.__name__,
                                   self.address, self.port, self._server_name)


class SniEndPointFactory(EndPointFactory):

    def __init__(self, proxy_address, port):
        self._proxy_address = proxy_address
        self._port = port

    def create(self, row):
        host_id = row.get("host_id")
        if host_id is None:
            raise ValueError("No host_id to create the SniEndPoint")

        return SniEndPoint(self._proxy_address, str(host_id), self._port)

    def create_from_sni(self, sni):
        return SniEndPoint(self._proxy_address, sni, self._port)


@total_ordering
class UnixSocketEndPoint(EndPoint):
    """
    Unix Socket EndPoint implementation.
    """

    def __init__(self, unix_socket_path):
        self._unix_socket_path = unix_socket_path

    @property
    def address(self):
        return self._unix_socket_path

    @property
    def port(self):
        return None

    @property
    def socket_family(self):
        return socket.AF_UNIX

    def resolve(self):
        return self.address, None

    def __eq__(self, other):
        return (isinstance(other, UnixSocketEndPoint) and
                self._unix_socket_path == other._unix_socket_path)

    def __hash__(self):
        return hash(self._unix_socket_path)

    def __lt__(self, other):
        return self._unix_socket_path < other._unix_socket_path

    def __str__(self):
        return str("%s" % (self._unix_socket_path,))

    def __repr__(self):
        return "<%s: %s>" % (self.__class__.__name__, self._unix_socket_path)


class _Frame(object):
    def __init__(self, version, flags, stream, opcode, body_offset, end_pos):
        self.version = version
        self.flags = flags
        self.stream = stream
        self.opcode = opcode
        self.body_offset = body_offset
        self.end_pos = end_pos

    def __eq__(self, other):  # facilitates testing
        if isinstance(other, _Frame):
            return (self.version == other.version and
                    self.flags == other.flags and
                    self.stream == other.stream and
                    self.opcode == other.opcode and
                    self.body_offset == other.body_offset and
                    self.end_pos == other.end_pos)
        return NotImplemented

    def __str__(self):
        return "ver({0}); flags({1:04b}); stream({2}); op({3}); offset({4}); len({5})".format(self.version, self.flags, self.stream, self.opcode, self.body_offset, self.end_pos - self.body_offset)


NONBLOCKING = (errno.EAGAIN, errno.EWOULDBLOCK)


class ConnectionException(Exception):
    """
    An unrecoverable error was hit when attempting to use a connection,
    or the connection was already closed or defunct.
    """

    def __init__(self, message, endpoint=None):
        Exception.__init__(self, message)
        self.endpoint = endpoint

    @property
    def host(self):
        return self.endpoint.address


class ConnectionShutdown(ConnectionException):
    """
    Raised when a connection has been marked as defunct or has been closed.
    """
    pass


class ProtocolVersionUnsupported(ConnectionException):
    """
    Server rejected startup message due to unsupported protocol version
    """
    def __init__(self, endpoint, startup_version):
        msg = "Unsupported protocol version on %s: %d" % (endpoint, startup_version)
        super(ProtocolVersionUnsupported, self).__init__(msg, endpoint)
        self.startup_version = startup_version


class ConnectionBusy(Exception):
    """
    An attempt was made to send a message through a :class:`.Connection` that
    was already at the max number of in-flight operations.
    """
    pass


class ProtocolError(Exception):
    """
    Communication did not match the protocol that this driver expects.
    """
    pass


class CrcMismatchException(ConnectionException):
    pass


class ContinuousPagingState(object):
    """
     A class for specifying continuous paging state, only supported starting with DSE_V2.
    """

    num_pages_requested = None
    """
    How many pages we have already requested
    """

    num_pages_received = None
    """
    How many pages we have already received
    """

    max_queue_size = None
    """
    The max queue size chosen by the user via the options
    """

    def __init__(self, max_queue_size):
        self.num_pages_requested = max_queue_size  # the initial query requests max_queue_size
        self.num_pages_received = 0
        self.max_queue_size = max_queue_size


class ContinuousPagingSession(object):
    def __init__(self, stream_id, decoder, row_factory, connection, state):
        self.stream_id = stream_id
        self.decoder = decoder
        self.row_factory = row_factory
        self.connection = connection
        self._condition = Condition()
        self._stop = False
        self._page_queue = deque()
        self._state = state
        self.released = False

    def on_message(self, result):
        if isinstance(result, ResultMessage):
            self.on_page(result)
        elif isinstance(result, ErrorMessage):
            self.on_error(result)

    def on_page(self, result):
        with self._condition:
            if self._state:
                self._state.num_pages_received += 1
            self._page_queue.appendleft((result.column_names, result.parsed_rows, None))
            self._stop |= result.continuous_paging_last
            self._condition.notify()

        if result.continuous_paging_last:
            self.released = True

    def on_error(self, error):
        if isinstance(error, ErrorMessage):
            error = error.to_exception()

        log.debug("Got error %s for session %s", error, self.stream_id)

        with self._condition:
            self._page_queue.appendleft((None, None, error))
            self._stop = True
            self._condition.notify()

        self.released = True

    def results(self):
        try:
            self._condition.acquire()
            while True:
                while not self._page_queue and not self._stop:
                    self._condition.wait(timeout=5)
                while self._page_queue:
                    names, rows, err = self._page_queue.pop()
                    if err:
                        raise err
                    self.maybe_request_more()
                    self._condition.release()
                    for row in self.row_factory(names, rows):
                        yield row
                    self._condition.acquire()
                if self._stop:
                    break
        finally:
            try:
                self._condition.release()
            except RuntimeError:
                # This exception happens if the CP results are not entirely consumed
                # and the session is terminated by the runtime. See PYTHON-1054
                pass

    def maybe_request_more(self):
        if not self._state:
            return

        max_queue_size = self._state.max_queue_size
        num_in_flight = self._state.num_pages_requested - self._state.num_pages_received
        space_in_queue = max_queue_size - len(self._page_queue) - num_in_flight
        log.debug("Session %s from %s, space in CP queue: %s, requested: %s, received: %s, num_in_flight: %s",
                  self.stream_id, self.connection.host, space_in_queue, self._state.num_pages_requested,
                  self._state.num_pages_received, num_in_flight)

        if space_in_queue >= max_queue_size / 2:
            self.update_next_pages(space_in_queue)

    def update_next_pages(self, num_next_pages):
        try:
            self._state.num_pages_requested += num_next_pages
            log.debug("Updating backpressure for session %s from %s", self.stream_id, self.connection.host)
            with self.connection.lock:
                self.connection.send_msg(ReviseRequestMessage(ReviseRequestMessage.RevisionType.PAGING_BACKPRESSURE,
                                                              self.stream_id,
                                                              next_pages=num_next_pages),
                                         self.connection.get_request_id(),
                                         self._on_backpressure_response)
        except ConnectionShutdown as ex:
            log.debug("Failed to update backpressure for session %s from %s, connection is shutdown",
                      self.stream_id, self.connection.host)
            self.on_error(ex)

    def _on_backpressure_response(self, response):
        if isinstance(response, ResultMessage):
            log.debug("Paging session %s backpressure updated.", self.stream_id)
        else:
            log.error("Failed updating backpressure for session %s from %s: %s", self.stream_id, self.connection.host,
                      response.to_exception() if hasattr(response, 'to_exception') else response)
            self.on_error(response)

    def cancel(self):
        try:
            log.debug("Canceling paging session %s from %s", self.stream_id, self.connection.host)
            with self.connection.lock:
                self.connection.send_msg(ReviseRequestMessage(ReviseRequestMessage.RevisionType.PAGING_CANCEL,
                                                              self.stream_id),
                                         self.connection.get_request_id(),
                                         self._on_cancel_response)
        except ConnectionShutdown:
            log.debug("Failed to cancel session %s from %s, connection is shutdown",
                      self.stream_id, self.connection.host)

        with self._condition:
            self._stop = True
            self._condition.notify()

    def _on_cancel_response(self, response):
        if isinstance(response, ResultMessage):
            log.debug("Paging session %s canceled.", self.stream_id)
        else:
            log.error("Failed canceling streaming session %s from %s: %s", self.stream_id, self.connection.host,
                      response.to_exception() if hasattr(response, 'to_exception') else response)
        self.released = True


def defunct_on_error(f):

    @wraps(f)
    def wrapper(self, *args, **kwargs):
        try:
            return f(self, *args, **kwargs)
        except Exception as exc:
            self.defunct(exc)
    return wrapper


DEFAULT_CQL_VERSION = '3.0.0'

if six.PY3:
    def int_from_buf_item(i):
        return i
else:
    int_from_buf_item = ord


class _ConnectionIOBuffer(object):
    """
    Abstraction class to ease the use of the different connection io buffers. With
    protocol V5 and checksumming, the data is read, validated and copied to another
    cql frame buffer.
    """
    _io_buffer = None
    _cql_frame_buffer = None
    _connection = None
    _segment_consumed = False

    def __init__(self, connection):
        self._io_buffer = io.BytesIO()
        self._connection = weakref.proxy(connection)

    @property
    def io_buffer(self):
        return self._io_buffer

    @property
    def cql_frame_buffer(self):
        return self._cql_frame_buffer if self.is_checksumming_enabled else \
            self._io_buffer

    def set_checksumming_buffer(self):
        self.reset_io_buffer()
        self._cql_frame_buffer = io.BytesIO()

    @property
    def is_checksumming_enabled(self):
        return self._connection._is_checksumming_enabled

    @property
    def has_consumed_segment(self):
        return self._segment_consumed;

    def readable_io_bytes(self):
        return self.io_buffer.tell()

    def readable_cql_frame_bytes(self):
        return self.cql_frame_buffer.tell()

    def reset_io_buffer(self):
        self._io_buffer = io.BytesIO(self._io_buffer.read())
        self._io_buffer.seek(0, 2)  # 2 == SEEK_END

    def reset_cql_frame_buffer(self):
        if self.is_checksumming_enabled:
            self._cql_frame_buffer = io.BytesIO(self._cql_frame_buffer.read())
            self._cql_frame_buffer.seek(0, 2)  # 2 == SEEK_END
        else:
            self.reset_io_buffer()


class Connection(object):

    CALLBACK_ERR_THREAD_THRESHOLD = 100

    in_buffer_size = 4096
    out_buffer_size = 4096

    cql_version = None
    no_compact = False
    protocol_version = ProtocolVersion.MAX_SUPPORTED

    keyspace = None
    compression = True
    _compression_type = None
    compressor = None
    decompressor = None

    endpoint = None
    ssl_options = None
    ssl_context = None
    last_error = None

    # The current number of operations that are in flight. More precisely,
    # the number of request IDs that are currently in use.
    # This includes orphaned requests.
    in_flight = 0

    # Max concurrent requests allowed per connection. This is set optimistically high, allowing
    # all request ids to be used in protocol version 3+. Normally concurrency would be controlled
    # at a higher level by the application or concurrent.execute_concurrent. This attribute
    # is for lower-level integrations that want some upper bound without reimplementing.
    max_in_flight = 2 ** 15

    # A set of available request IDs.  When using the v3 protocol or higher,
    # this will not initially include all request IDs in order to save memory,
    # but the set will grow if it is exhausted.
    request_ids = None

    # Tracks the highest used request ID in order to help with growing the
    # request_ids set
    highest_request_id = 0

    # Tracks the request IDs which are no longer waited on (timed out), but
    # cannot be reused yet because the node might still send a response
    # on this stream
    orphaned_request_ids = None

    # Set to true if the orphaned stream ID count cross configured threshold
    # and the connection will be replaced
    orphaned_threshold_reached = False

    # If the number of orphaned streams reaches this threshold, this connection
    # will become marked and will be replaced with a new connection by the
    # owning pool (currently, only HostConnection supports this)
    orphaned_threshold = 3  * max_in_flight // 4

    is_defunct = False
    is_closed = False
    lock = None
    user_type_map = None

    msg_received = False

    is_unsupported_proto_version = False

    is_control_connection = False
    signaled_error = False  # used for flagging at the pool level

    allow_beta_protocol_version = False

    _current_frame = None

    _socket = None

    _socket_impl = socket
    _ssl_impl = ssl

    _check_hostname = False
    _product_type = None

    _is_checksumming_enabled = False

    _on_orphaned_stream_released = None

    @property
    def _iobuf(self):
        # backward compatibility, to avoid any change in the reactors
        return self._io_buffer.io_buffer

    def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
                 ssl_options=None, sockopts=None, compression=True,
                 cql_version=None, protocol_version=ProtocolVersion.MAX_SUPPORTED, is_control_connection=False,
                 user_type_map=None, connect_timeout=None, allow_beta_protocol_version=False, no_compact=False,
                 ssl_context=None, on_orphaned_stream_released=None):

        # TODO next major rename host to endpoint and remove port kwarg.
        self.endpoint = host if isinstance(host, EndPoint) else DefaultEndPoint(host, port)

        self.authenticator = authenticator
        self.ssl_options = ssl_options.copy() if ssl_options else None
        self.ssl_context = ssl_context
        self.sockopts = sockopts
        self.compression = compression
        self.cql_version = cql_version
        self.protocol_version = protocol_version
        self.is_control_connection = is_control_connection
        self.user_type_map = user_type_map
        self.connect_timeout = connect_timeout
        self.allow_beta_protocol_version = allow_beta_protocol_version
        self.no_compact = no_compact
        self._push_watchers = defaultdict(set)
        self._requests = {}
        self._io_buffer = _ConnectionIOBuffer(self)
        self._continuous_paging_sessions = {}
        self._socket_writable = True
        self.orphaned_request_ids = set()
        self._on_orphaned_stream_released = on_orphaned_stream_released

        if ssl_options:
            self._check_hostname = bool(self.ssl_options.pop('check_hostname', False))
            if self._check_hostname:
                if not getattr(ssl, 'match_hostname', None):
                    raise RuntimeError("ssl_options specify 'check_hostname', but ssl.match_hostname is not provided. "
                                       "Patch or upgrade Python to use this option.")
            self.ssl_options.update(self.endpoint.ssl_options or {})
        elif self.endpoint.ssl_options:
            self.ssl_options = self.endpoint.ssl_options


        if protocol_version >= 3:
            self.max_request_id = min(self.max_in_flight - 1, (2 ** 15) - 1)
            # Don't fill the deque with 2**15 items right away. Start with some and add
            # more if needed.
            initial_size = min(300, self.max_in_flight)
            self.request_ids = deque(range(initial_size))
            self.highest_request_id = initial_size - 1
        else:
            self.max_request_id = min(self.max_in_flight, (2 ** 7) - 1)
            self.request_ids = deque(range(self.max_request_id + 1))
            self.highest_request_id = self.max_request_id

        self.lock = RLock()
        self.connected_event = Event()

    @property
    def host(self):
        return self.endpoint.address

    @property
    def port(self):
        return self.endpoint.port

    @classmethod
    def initialize_reactor(cls):
        """
        Called once by Cluster.connect().  This should be used by implementations
        to set up any resources that will be shared across connections.
        """
        pass

    @classmethod
    def handle_fork(cls):
        """
        Called after a forking.  This should cleanup any remaining reactor state
        from the parent process.
        """
        pass

    @classmethod
    def create_timer(cls, timeout, callback):
        raise NotImplementedError()

    @classmethod
    def factory(cls, endpoint, timeout, *args, **kwargs):
        """
        A factory function which returns connections which have
        succeeded in connecting and are ready for service (or
        raises an exception otherwise).
        """
        start = time.time()
        kwargs['connect_timeout'] = timeout
        conn = cls(endpoint, *args, **kwargs)
        elapsed = time.time() - start
        conn.connected_event.wait(timeout - elapsed)
        if conn.last_error:
            if conn.is_unsupported_proto_version:
                raise ProtocolVersionUnsupported(endpoint, conn.protocol_version)
            raise conn.last_error
        elif not conn.connected_event.is_set():
            conn.close()
            raise OperationTimedOut("Timed out creating connection (%s seconds)" % timeout)
        else:
            return conn

    def _wrap_socket_from_context(self):
        ssl_options = self.ssl_options or {}
        # PYTHON-1186: set the server_hostname only if the SSLContext has
        # check_hostname enabled and it is not already provided by the EndPoint ssl options
        if (self.ssl_context.check_hostname and
                'server_hostname' not in ssl_options):
            ssl_options = ssl_options.copy()
            ssl_options['server_hostname'] = self.endpoint.address
        self._socket = self.ssl_context.wrap_socket(self._socket, **ssl_options)

    def _initiate_connection(self, sockaddr):
        self._socket.connect(sockaddr)

    def _match_hostname(self):
        ssl.match_hostname(self._socket.getpeercert(), self.endpoint.address)

    def _get_socket_addresses(self):
        address, port = self.endpoint.resolve()

        if hasattr(socket, 'AF_UNIX') and self.endpoint.socket_family == socket.AF_UNIX:
            return [(socket.AF_UNIX, socket.SOCK_STREAM, 0, None, address)]

        addresses = socket.getaddrinfo(address, port, self.endpoint.socket_family, socket.SOCK_STREAM)
        if not addresses:
            raise ConnectionException("getaddrinfo returned empty list for %s" % (self.endpoint,))

        return addresses

    def _connect_socket(self):
        sockerr = None
        addresses = self._get_socket_addresses()
        for (af, socktype, proto, _, sockaddr) in addresses:
            try:
                self._socket = self._socket_impl.socket(af, socktype, proto)
                if self.ssl_context:
                    self._wrap_socket_from_context()
                elif self.ssl_options:
                    if not self._ssl_impl:
                        raise RuntimeError("This version of Python was not compiled with SSL support")
                    self._socket = self._ssl_impl.wrap_socket(self._socket, **self.ssl_options)
                self._socket.settimeout(self.connect_timeout)
                self._initiate_connection(sockaddr)
                self._socket.settimeout(None)
                if self._check_hostname:
                    self._match_hostname()
                sockerr = None
                break
            except socket.error as err:
                if self._socket:
                    self._socket.close()
                    self._socket = None
                sockerr = err

        if sockerr:
            raise socket.error(sockerr.errno, "Tried connecting to %s. Last error: %s" %
                               ([a[4] for a in addresses], sockerr.strerror or sockerr))

        if self.sockopts:
            for args in self.sockopts:
                self._socket.setsockopt(*args)

    def _enable_compression(self):
        if self._compressor:
            self.compressor = self._compressor

    def _enable_checksumming(self):
        self._io_buffer.set_checksumming_buffer()
        self._is_checksumming_enabled = True
        self._segment_codec = segment_codec_lz4 if self.compressor else segment_codec_no_compression
        log.debug("Enabling protocol checksumming on connection (%s).", id(self))

    def close(self):
        raise NotImplementedError()

    def defunct(self, exc):
        with self.lock:
            if self.is_defunct or self.is_closed:
                return
            self.is_defunct = True

        exc_info = sys.exc_info()
        # if we are not handling an exception, just use the passed exception, and don't try to format exc_info with the message
        if any(exc_info):
            log.debug("Defuncting connection (%s) to %s:",
                      id(self), self.endpoint, exc_info=exc_info)
        else:
            log.debug("Defuncting connection (%s) to %s: %s",
                      id(self), self.endpoint, exc)

        self.last_error = exc
        self.close()
        self.error_all_cp_sessions(exc)
        self.error_all_requests(exc)
        self.connected_event.set()
        return exc

    def error_all_cp_sessions(self, exc):
        stream_ids = list(self._continuous_paging_sessions.keys())
        for stream_id in stream_ids:
            self._continuous_paging_sessions[stream_id].on_error(exc)

    def error_all_requests(self, exc):
        with self.lock:
            requests = self._requests
            self._requests = {}

        if not requests:
            return

        new_exc = ConnectionShutdown(str(exc))

        def try_callback(cb):
            try:
                cb(new_exc)
            except Exception:
                log.warning("Ignoring unhandled exception while erroring requests for a "
                            "failed connection (%s) to host %s:",
                            id(self), self.endpoint, exc_info=True)

        # run first callback from this thread to ensure pool state before leaving
        cb, _, _ = requests.popitem()[1]
        try_callback(cb)

        if not requests:
            return

        # additional requests are optionally errored from a separate thread
        # The default callback and retry logic is fairly expensive -- we don't
        # want to tie up the event thread when there are many requests
        def err_all_callbacks():
            for cb, _, _ in requests.values():
                try_callback(cb)
        if len(requests) < Connection.CALLBACK_ERR_THREAD_THRESHOLD:
            err_all_callbacks()
        else:
            # daemon thread here because we want to stay decoupled from the cluster TPE
            # TODO: would it make sense to just have a driver-global TPE?
            t = Thread(target=err_all_callbacks)