@@ -701,14 +699,6 @@ pipeline {
| LIBEV |
A full-featured and high-performance event loop that is loosely modeled after libevent, but without its limitations and bugs |
-
- | GEVENT |
- A co-routine -based Python networking library that uses greenlet to provide a high-level synchronous API on top of the libev or libuv event loop |
-
-
- | EVENTLET |
- A concurrent networking library for Python that allows you to change how you run your code, not how you write it |
-
| ASYNCIO |
A library to write concurrent code using the async/await syntax |
@@ -717,10 +707,6 @@ pipeline {
ASYNCORE |
A module provides the basic infrastructure for writing asynchronous socket service clients and servers |
-
- | TWISTED |
- An event-driven networking engine written in Python and licensed under the open source MIT license |
-
''')
choice(
name: 'CI_SCHEDULE',
diff --git a/benchmarks/base.py b/benchmarks/base.py
index 290ba28788..bdfcfa4493 100644
--- a/benchmarks/base.py
+++ b/benchmarks/base.py
@@ -67,15 +67,6 @@
except (ImportError, SyntaxError):
pass
-have_twisted = False
-try:
- from cassandra.io.twistedreactor import TwistedConnection
- have_twisted = True
- supported_reactors.append(TwistedConnection)
-except ImportError as exc:
- log.exception("Error importing twisted")
- pass
-
KEYSPACE = "testkeyspace" + str(int(time.time()))
TABLE = "testtable"
@@ -230,8 +221,6 @@ def parse_options():
help='only benchmark with asyncio connections')
parser.add_option('--libev-only', action='store_true', dest='libev_only',
help='only benchmark with libev connections')
- parser.add_option('--twisted-only', action='store_true', dest='twisted_only',
- help='only benchmark with Twisted connections')
parser.add_option('-m', '--metrics', action='store_true', dest='enable_metrics',
help='enable and print metrics for operations')
parser.add_option('-l', '--log-level', default='info',
@@ -271,11 +260,6 @@ def parse_options():
log.error("libev is not available")
sys.exit(1)
options.supported_reactors = [LibevConnection]
- elif options.twisted_only:
- if not have_twisted:
- log.error("Twisted is not available")
- sys.exit(1)
- options.supported_reactors = [TwistedConnection]
else:
options.supported_reactors = supported_reactors
if not have_libev:
diff --git a/cassandra/cluster.py b/cassandra/cluster.py
index 6b2ab4b288..9ffefe99b5 100644
--- a/cassandra/cluster.py
+++ b/cassandra/cluster.py
@@ -93,55 +93,11 @@
from cassandra.datastax.graph.query import _request_timeout_key, _GraphSONContextRowFactory
from cassandra.datastax import cloud as dscloud
-try:
- from cassandra.io.twistedreactor import TwistedConnection
-except ImportError:
- TwistedConnection = None
-
-try:
- from cassandra.io.eventletreactor import EventletConnection
-# PYTHON-1364
-#
-# At the moment eventlet initialization is chucking AttributeErrors due to its dependence on pyOpenSSL
-# and some changes in Python 3.12 which have some knock-on effects there.
-except (ImportError, AttributeError):
- EventletConnection = None
-
try:
from weakref import WeakSet
except ImportError:
from cassandra.util import WeakSet # NOQA
-def _is_gevent_monkey_patched():
- if 'gevent.monkey' not in sys.modules:
- return False
- import gevent.socket
- return socket.socket is gevent.socket.socket
-
-def _try_gevent_import():
- if _is_gevent_monkey_patched():
- from cassandra.io.geventreactor import GeventConnection
- return (GeventConnection,None)
- else:
- return (None,None)
-
-def _is_eventlet_monkey_patched():
- if 'eventlet.patcher' not in sys.modules:
- return False
- try:
- import eventlet.patcher
- return eventlet.patcher.is_monkey_patched('socket')
- # Another case related to PYTHON-1364
- except AttributeError:
- return False
-
-def _try_eventlet_import():
- if _is_eventlet_monkey_patched():
- from cassandra.io.eventletreactor import EventletConnection
- return (EventletConnection,None)
- else:
- return (None,None)
-
def _try_libev_import():
try:
from cassandra.io.libevreactor import LibevConnection
@@ -168,7 +124,7 @@ def _connection_reduce_fn(val,import_fn):
log = logging.getLogger(__name__)
-conn_fns = (_try_gevent_import, _try_eventlet_import, _try_libev_import, _try_asyncore_import)
+conn_fns = (_try_libev_import, _try_asyncore_import)
(conn_class, excs) = reduce(_connection_reduce_fn, conn_fns, (None,[]))
if not conn_class:
raise DependencyException("Unable to load a default connection class", excs)
@@ -878,9 +834,6 @@ def default_retry_policy(self, policy):
* :class:`cassandra.io.asyncorereactor.AsyncoreConnection`
* :class:`cassandra.io.libevreactor.LibevConnection`
- * :class:`cassandra.io.eventletreactor.EventletConnection` (requires monkey-patching - see doc for details)
- * :class:`cassandra.io.geventreactor.GeventConnection` (requires monkey-patching - see doc for details)
- * :class:`cassandra.io.twistedreactor.TwistedConnection`
* EXPERIMENTAL: :class:`cassandra.io.asyncioreactor.AsyncioConnection`
By default, ``AsyncoreConnection`` will be used, which uses
@@ -888,9 +841,6 @@ def default_retry_policy(self, policy):
If ``libev`` is installed, ``LibevConnection`` will be used instead.
- If ``gevent`` or ``eventlet`` monkey-patching is detected, the corresponding
- connection class will be used automatically.
-
``AsyncioConnection``, which uses the ``asyncio`` module in the Python
standard library, is also available, but currently experimental. Note that
it requires ``asyncio`` features that were only introduced in the 3.4 line
@@ -1168,9 +1118,7 @@ def __init__(self,
raise ValueError("contact_points, endpoint_factory, ssl_context, and ssl_options "
"cannot be specified with a cloud configuration")
- uses_twisted = TwistedConnection and issubclass(self.connection_class, TwistedConnection)
- uses_eventlet = EventletConnection and issubclass(self.connection_class, EventletConnection)
- cloud_config = dscloud.get_cloud_config(cloud, create_pyopenssl_context=uses_twisted or uses_eventlet)
+ cloud_config = dscloud.get_cloud_config(cloud)
ssl_context = cloud_config.ssl_context
ssl_options = {'check_hostname': True}
@@ -1389,7 +1337,7 @@ def __init__(self,
HostDistance.REMOTE: DEFAULT_MAX_CONNECTIONS_PER_REMOTE_HOST
}
- self.executor = self._create_thread_pool_executor(max_workers=executor_threads)
+ self.executor = ThreadPoolExecutor(max_workers=executor_threads)
self.scheduler = _Scheduler(self.executor)
self._lock = RLock()
@@ -1411,42 +1359,6 @@ def __init__(self,
if application_version is not None:
self.application_version = application_version
- def _create_thread_pool_executor(self, **kwargs):
- """
- Create a ThreadPoolExecutor for the cluster. In most cases, the built-in
- `concurrent.futures.ThreadPoolExecutor` is used.
-
- Python 3.7+ and Eventlet cause the `concurrent.futures.ThreadPoolExecutor`
- to hang indefinitely. In that case, the user needs to have the `futurist`
- package so we can use the `futurist.GreenThreadPoolExecutor` class instead.
-
- :param kwargs: All keyword args are passed to the ThreadPoolExecutor constructor.
- :return: A ThreadPoolExecutor instance.
- """
- tpe_class = ThreadPoolExecutor
- if sys.version_info[0] >= 3 and sys.version_info[1] >= 7:
- try:
- from cassandra.io.eventletreactor import EventletConnection
- is_eventlet = issubclass(self.connection_class, EventletConnection)
- except:
- # Eventlet is not available or can't be detected
- return tpe_class(**kwargs)
-
- if is_eventlet:
- try:
- from futurist import GreenThreadPoolExecutor
- tpe_class = GreenThreadPoolExecutor
- except ImportError:
- # futurist is not available
- raise ImportError(
- ("Python 3.7+ and Eventlet cause the `concurrent.futures.ThreadPoolExecutor` "
- "to hang indefinitely. If you want to use the Eventlet reactor, you "
- "need to install the `futurist` package to allow the driver to use "
- "the GreenThreadPoolExecutor. See https://github.com/eventlet/eventlet/issues/508 "
- "for more details."))
-
- return tpe_class(**kwargs)
-
def register_user_type(self, keyspace, user_type, klass):
"""
Registers a class to use to represent a particular user-defined type.
diff --git a/cassandra/connection.py b/cassandra/connection.py
index 3ceaa08afc..48aa709821 100644
--- a/cassandra/connection.py
+++ b/cassandra/connection.py
@@ -29,10 +29,7 @@
import weakref
-if 'gevent.monkey' in sys.modules:
- from gevent.queue import Queue, Empty
-else:
- from queue import Queue, Empty # noqa
+from queue import Queue, Empty # noqa
from cassandra import ConsistencyLevel, AuthenticationFailed, OperationTimedOut, ProtocolVersion
from cassandra.marshal import int32_pack
diff --git a/cassandra/datastax/cloud/__init__.py b/cassandra/datastax/cloud/__init__.py
index e175b2928b..f5633b5e62 100644
--- a/cassandra/datastax/cloud/__init__.py
+++ b/cassandra/datastax/cloud/__init__.py
@@ -78,7 +78,7 @@ def from_dict(cls, d):
return c
-def get_cloud_config(cloud_config, create_pyopenssl_context=False):
+def get_cloud_config(cloud_config):
if not _HAS_SSL:
raise DriverException("A Python installation with SSL is required to connect to a cloud cluster.")
@@ -86,17 +86,15 @@ def get_cloud_config(cloud_config, create_pyopenssl_context=False):
raise ValueError("The cloud config doesn't have a secure_connect_bundle specified.")
try:
- config = read_cloud_config_from_zip(cloud_config, create_pyopenssl_context)
+ config = read_cloud_config_from_zip(cloud_config)
except BadZipFile:
raise ValueError("Unable to open the zip file for the cloud config. Check your secure connect bundle.")
config = read_metadata_info(config, cloud_config)
- if create_pyopenssl_context:
- config.ssl_context = config.pyopenssl_context
return config
-def read_cloud_config_from_zip(cloud_config, create_pyopenssl_context):
+def read_cloud_config_from_zip(cloud_config):
secure_bundle = cloud_config['secure_connect_bundle']
use_default_tempdir = cloud_config.get('use_default_tempdir', None)
with ZipFile(secure_bundle) as zipfile:
@@ -104,12 +102,12 @@ def read_cloud_config_from_zip(cloud_config, create_pyopenssl_context):
tmp_dir = tempfile.mkdtemp(dir=base_dir)
try:
zipfile.extractall(path=tmp_dir)
- return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config, create_pyopenssl_context)
+ return parse_cloud_config(os.path.join(tmp_dir, 'config.json'), cloud_config)
finally:
shutil.rmtree(tmp_dir)
-def parse_cloud_config(path, cloud_config, create_pyopenssl_context):
+def parse_cloud_config(path, cloud_config):
with open(path, 'r') as stream:
data = json.load(stream)
@@ -123,11 +121,7 @@ def parse_cloud_config(path, cloud_config, create_pyopenssl_context):
ca_cert_location = os.path.join(config_dir, 'ca.crt')
cert_location = os.path.join(config_dir, 'cert')
key_location = os.path.join(config_dir, 'key')
- # Regardless of if we create a pyopenssl context, we still need the builtin one
- # to connect to the metadata service
config.ssl_context = _ssl_context_from_cert(ca_cert_location, cert_location, key_location)
- if create_pyopenssl_context:
- config.pyopenssl_context = _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location)
return config
@@ -178,18 +172,3 @@ def _ssl_context_from_cert(ca_cert_location, cert_location, key_location):
return ssl_context
-
-def _pyopenssl_context_from_cert(ca_cert_location, cert_location, key_location):
- try:
- from OpenSSL import SSL
- except ImportError as e:
- raise ImportError(
- "PyOpenSSL must be installed to connect to Astra with the Eventlet or Twisted event loops")\
- .with_traceback(e.__traceback__)
- ssl_context = SSL.Context(SSL.TLSv1_METHOD)
- ssl_context.set_verify(SSL.VERIFY_PEER, callback=lambda _1, _2, _3, _4, ok: ok)
- ssl_context.use_certificate_file(cert_location)
- ssl_context.use_privatekey_file(key_location)
- ssl_context.load_verify_locations(ca_cert_location)
-
- return ssl_context
\ No newline at end of file
diff --git a/cassandra/datastax/insights/reporter.py b/cassandra/datastax/insights/reporter.py
index 607c723a1a..e3ea5a1c3a 100644
--- a/cassandra/datastax/insights/reporter.py
+++ b/cassandra/datastax/insights/reporter.py
@@ -144,11 +144,7 @@ def _get_startup_data(self):
cert_validation = None
try:
if self._session.cluster.ssl_context:
- if isinstance(self._session.cluster.ssl_context, ssl.SSLContext):
- cert_validation = self._session.cluster.ssl_context.verify_mode == ssl.CERT_REQUIRED
- else: # pyopenssl
- from OpenSSL import SSL
- cert_validation = self._session.cluster.ssl_context.get_verify_mode() != SSL.VERIFY_NONE
+ cert_validation = self._session.cluster.ssl_context.verify_mode == ssl.CERT_REQUIRED
elif self._session.cluster.ssl_options:
cert_validation = self._session.cluster.ssl_options.get('cert_reqs') == ssl.CERT_REQUIRED
except Exception as e:
diff --git a/cassandra/io/eventletreactor.py b/cassandra/io/eventletreactor.py
deleted file mode 100644
index 6be7738236..0000000000
--- a/cassandra/io/eventletreactor.py
+++ /dev/null
@@ -1,195 +0,0 @@
-# Copyright 2014 Symantec Corporation
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you 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.
-
-# Originally derived from MagnetoDB source:
-# https://github.com/stackforge/magnetodb/blob/2015.1.0b1/magnetodb/common/cassandra/io/eventletreactor.py
-import eventlet
-from eventlet.green import socket
-from eventlet.queue import Queue
-from greenlet import GreenletExit
-import logging
-from threading import Event
-import time
-from deprecated import deprecated
-
-from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager
-try:
- from eventlet.green.OpenSSL import SSL
- _PYOPENSSL = True
-except ImportError as e:
- _PYOPENSSL = False
- no_pyopenssl_error = e
-
-
-log = logging.getLogger(__name__)
-
-
-def _check_pyopenssl():
- if not _PYOPENSSL:
- raise ImportError(
- "{}, pyOpenSSL must be installed to enable "
- "SSL support with the Eventlet event loop".format(str(no_pyopenssl_error))
- )
-
-
-@deprecated(version="3.30.0", reason="The eventlet event loop is deprecated and will be removed in 3.31.0. See CASSPYTHON-12.")
-class EventletConnection(Connection):
- """
- An implementation of :class:`.Connection` that utilizes ``eventlet``.
-
- This implementation assumes all eventlet monkey patching is active. It is not tested with partial patching.
- """
-
- _read_watcher = None
- _write_watcher = None
-
- _socket_impl = eventlet.green.socket
- _ssl_impl = eventlet.green.ssl
-
- _timers = None
- _timeout_watcher = None
- _new_timer = None
-
- @classmethod
- def initialize_reactor(cls):
- eventlet.monkey_patch()
- if not cls._timers:
- cls._timers = TimerManager()
- cls._timeout_watcher = eventlet.spawn(cls.service_timeouts)
- cls._new_timer = Event()
-
- @classmethod
- def create_timer(cls, timeout, callback):
- timer = Timer(timeout, callback)
- cls._timers.add_timer(timer)
- cls._new_timer.set()
- return timer
-
- @classmethod
- def service_timeouts(cls):
- """
- cls._timeout_watcher runs in this loop forever.
- It is usually waiting for the next timeout on the cls._new_timer Event.
- When new timers are added, that event is set so that the watcher can
- wake up and possibly set an earlier timeout.
- """
- timer_manager = cls._timers
- while True:
- next_end = timer_manager.service_timeouts()
- sleep_time = max(next_end - time.time(), 0) if next_end else 10000
- cls._new_timer.wait(sleep_time)
- cls._new_timer.clear()
-
- def __init__(self, *args, **kwargs):
- Connection.__init__(self, *args, **kwargs)
- self.uses_legacy_ssl_options = self.ssl_options and not self.ssl_context
- self._write_queue = Queue()
-
- self._connect_socket()
-
- self._read_watcher = eventlet.spawn(lambda: self.handle_read())
- self._write_watcher = eventlet.spawn(lambda: self.handle_write())
- self._send_options_message()
-
- def _wrap_socket_from_context(self):
- _check_pyopenssl()
- rv = SSL.Connection(self.ssl_context, self._socket)
- rv.set_connect_state()
- if self.ssl_options and 'server_hostname' in self.ssl_options:
- # This is necessary for SNI
- rv.set_tlsext_host_name(self.ssl_options['server_hostname'].encode('ascii'))
- return rv
-
- def _initiate_connection(self, sockaddr):
- if self.uses_legacy_ssl_options:
- super(EventletConnection, self)._initiate_connection(sockaddr)
- else:
- self._socket.connect(sockaddr)
- if self.ssl_context or self.ssl_options:
- self._socket.do_handshake()
-
- def _validate_hostname(self):
- if not self.uses_legacy_ssl_options:
- cert_name = self._socket.get_peer_certificate().get_subject().commonName
- if cert_name != self.endpoint.address:
- raise Exception("Hostname verification failed! Certificate name '{}' "
- "doesn't match endpoint '{}'".format(cert_name, self.endpoint.address))
-
- def close(self):
- with self.lock:
- if self.is_closed:
- return
- self.is_closed = True
-
- log.debug("Closing connection (%s) to %s" % (id(self), self.endpoint))
-
- cur_gthread = eventlet.getcurrent()
-
- if self._read_watcher and self._read_watcher != cur_gthread:
- self._read_watcher.kill()
- if self._write_watcher and self._write_watcher != cur_gthread:
- self._write_watcher.kill()
- if self._socket:
- self._socket.close()
- log.debug("Closed socket to %s" % (self.endpoint,))
-
- if not self.is_defunct:
- self.error_all_requests(
- ConnectionShutdown("Connection to %s was closed" % self.endpoint))
- # don't leave in-progress operations hanging
- self.connected_event.set()
-
- def handle_close(self):
- log.debug("connection closed by server")
- self.close()
-
- def handle_write(self):
- while True:
- try:
- next_msg = self._write_queue.get()
- self._socket.sendall(next_msg)
- except socket.error as err:
- log.debug("Exception during socket send for %s: %s", self, err)
- self.defunct(err)
- return # Leave the write loop
- except GreenletExit: # graceful greenthread exit
- return
-
- def handle_read(self):
- while True:
- try:
- buf = self._socket.recv(self.in_buffer_size)
- self._iobuf.write(buf)
- except socket.error as err:
- log.debug("Exception during socket recv for %s: %s",
- self, err)
- self.defunct(err)
- return # leave the read loop
- except GreenletExit: # graceful greenthread exit
- return
-
- if buf and self._iobuf.tell():
- self.process_io_buffer()
- else:
- log.debug("Connection %s closed by server", self)
- self.close()
- return
-
- def push(self, data):
- chunk_size = self.out_buffer_size
- for i in range(0, len(data), chunk_size):
- self._write_queue.put(data[i:i + chunk_size])
diff --git a/cassandra/io/geventreactor.py b/cassandra/io/geventreactor.py
deleted file mode 100644
index eb1296d6f9..0000000000
--- a/cassandra/io/geventreactor.py
+++ /dev/null
@@ -1,138 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you 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.
-import gevent
-import gevent.event
-from gevent.queue import Queue
-from gevent import socket
-import gevent.ssl
-
-from deprecated import deprecated
-import logging
-import time
-
-from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager
-
-log = logging.getLogger(__name__)
-
-@deprecated(version="3.30.0", reason="The gevent event loop is deprecated and will be removed in 3.31.0. See CASSPYTHON-12.")
-class GeventConnection(Connection):
- """
- An implementation of :class:`.Connection` that utilizes ``gevent``.
-
- This implementation assumes all gevent monkey patching is active. It is not tested with partial patching.
- """
-
- _read_watcher = None
- _write_watcher = None
-
- _socket_impl = gevent.socket
- _ssl_impl = gevent.ssl
-
- _timers = None
- _timeout_watcher = None
- _new_timer = None
-
- @classmethod
- def initialize_reactor(cls):
- if not cls._timers:
- cls._timers = TimerManager()
- cls._timeout_watcher = gevent.spawn(cls.service_timeouts)
- cls._new_timer = gevent.event.Event()
-
- @classmethod
- def create_timer(cls, timeout, callback):
- timer = Timer(timeout, callback)
- cls._timers.add_timer(timer)
- cls._new_timer.set()
- return timer
-
- @classmethod
- def service_timeouts(cls):
- timer_manager = cls._timers
- timer_event = cls._new_timer
- while True:
- next_end = timer_manager.service_timeouts()
- sleep_time = max(next_end - time.time(), 0) if next_end else 10000
- timer_event.wait(sleep_time)
- timer_event.clear()
-
- def __init__(self, *args, **kwargs):
- Connection.__init__(self, *args, **kwargs)
-
- self._write_queue = Queue()
-
- self._connect_socket()
-
- self._read_watcher = gevent.spawn(self.handle_read)
- self._write_watcher = gevent.spawn(self.handle_write)
- self._send_options_message()
-
- def close(self):
- with self.lock:
- if self.is_closed:
- return
- self.is_closed = True
-
- log.debug("Closing connection (%s) to %s" % (id(self), self.endpoint))
- if self._read_watcher:
- self._read_watcher.kill(block=False)
- if self._write_watcher:
- self._write_watcher.kill(block=False)
- if self._socket:
- self._socket.close()
- log.debug("Closed socket to %s" % (self.endpoint,))
-
- if not self.is_defunct:
- self.error_all_requests(
- ConnectionShutdown("Connection to %s was closed" % self.endpoint))
- # don't leave in-progress operations hanging
- self.connected_event.set()
-
- def handle_close(self):
- log.debug("connection closed by server")
- self.close()
-
- def handle_write(self):
- while True:
- try:
- next_msg = self._write_queue.get()
- self._socket.sendall(next_msg)
- except socket.error as err:
- log.debug("Exception in send for %s: %s", self, err)
- self.defunct(err)
- return
-
- def handle_read(self):
- while True:
- try:
- buf = self._socket.recv(self.in_buffer_size)
- self._iobuf.write(buf)
- except socket.error as err:
- log.debug("Exception in read for %s: %s", self, err)
- self.defunct(err)
- return # leave the read loop
-
- if buf and self._iobuf.tell():
- self.process_io_buffer()
- else:
- log.debug("Connection %s closed by server", self)
- self.close()
- return
-
- def push(self, data):
- chunk_size = self.out_buffer_size
- for i in range(0, len(data), chunk_size):
- self._write_queue.put(data[i:i + chunk_size])
diff --git a/cassandra/io/twistedreactor.py b/cassandra/io/twistedreactor.py
deleted file mode 100644
index 58e79e9ce9..0000000000
--- a/cassandra/io/twistedreactor.py
+++ /dev/null
@@ -1,309 +0,0 @@
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# to you 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.
-"""
-Module that implements an event loop based on twisted
-( https://twistedmatrix.com ).
-"""
-import atexit
-from deprecated import deprecated
-import logging
-import time
-from functools import partial
-from threading import Thread, Lock
-import weakref
-
-from twisted.internet import reactor, protocol
-from twisted.internet.endpoints import connectProtocol, TCP4ClientEndpoint, SSL4ClientEndpoint
-from twisted.internet.interfaces import IOpenSSLClientConnectionCreator
-from twisted.python.failure import Failure
-from zope.interface import implementer
-
-from cassandra.connection import Connection, ConnectionShutdown, Timer, TimerManager, ConnectionException
-
-try:
- from OpenSSL import SSL
- _HAS_SSL = True
-except ImportError as e:
- _HAS_SSL = False
- import_exception = e
-log = logging.getLogger(__name__)
-
-
-def _cleanup(cleanup_weakref):
- try:
- cleanup_weakref()._cleanup()
- except ReferenceError:
- return
-
-
-class TwistedConnectionProtocol(protocol.Protocol):
- """
- Twisted Protocol class for handling data received and connection
- made events.
- """
-
- def __init__(self, connection):
- self.connection = connection
-
- def dataReceived(self, data):
- """
- Callback function that is called when data has been received
- on the connection.
-
- Reaches back to the Connection object and queues the data for
- processing.
- """
- self.connection._iobuf.write(data)
- self.connection.handle_read()
-
- def connectionMade(self):
- """
- Callback function that is called when a connection has succeeded.
-
- Reaches back to the Connection object and confirms that the connection
- is ready.
- """
- self.connection.client_connection_made(self.transport)
-
- def connectionLost(self, reason):
- # reason is a Failure instance
- log.debug("Connect lost: %s", reason)
- self.connection.defunct(reason.value)
-
-
-class TwistedLoop(object):
-
- _lock = None
- _thread = None
- _timeout_task = None
- _timeout = None
-
- def __init__(self):
- self._lock = Lock()
- self._timers = TimerManager()
-
- def maybe_start(self):
- with self._lock:
- if not reactor.running:
- self._thread = Thread(target=reactor.run,
- name="cassandra_driver_twisted_event_loop",
- kwargs={'installSignalHandlers': False})
- self._thread.daemon = True
- self._thread.start()
- atexit.register(partial(_cleanup, weakref.ref(self)))
-
- def _reactor_stopped(self):
- return reactor._stopped
-
- def _cleanup(self):
- if self._thread:
- reactor.callFromThread(reactor.stop)
- self._thread.join(timeout=1.0)
- if self._thread.is_alive():
- log.warning("Event loop thread could not be joined, so "
- "shutdown may not be clean. Please call "
- "Cluster.shutdown() to avoid this.")
- log.debug("Event loop thread was joined")
-
- def add_timer(self, timer):
- self._timers.add_timer(timer)
- # callFromThread to schedule from the loop thread, where
- # the timeout task can safely be modified
- reactor.callFromThread(self._schedule_timeout, timer.end)
-
- def _schedule_timeout(self, next_timeout):
- if next_timeout:
- delay = max(next_timeout - time.time(), 0)
- if self._timeout_task and self._timeout_task.active():
- if next_timeout < self._timeout:
- self._timeout_task.reset(delay)
- self._timeout = next_timeout
- else:
- self._timeout_task = reactor.callLater(delay, self._on_loop_timer)
- self._timeout = next_timeout
-
- def _on_loop_timer(self):
- self._timers.service_timeouts()
- self._schedule_timeout(self._timers.next_timeout)
-
-
-@implementer(IOpenSSLClientConnectionCreator)
-class _SSLCreator(object):
- def __init__(self, endpoint, ssl_context, ssl_options, check_hostname, timeout):
- self.endpoint = endpoint
- self.ssl_options = ssl_options
- self.check_hostname = check_hostname
- self.timeout = timeout
-
- if ssl_context:
- self.context = ssl_context
- else:
- self.context = SSL.Context(SSL.TLSv1_METHOD)
- if "certfile" in self.ssl_options:
- self.context.use_certificate_file(self.ssl_options["certfile"])
- if "keyfile" in self.ssl_options:
- self.context.use_privatekey_file(self.ssl_options["keyfile"])
- if "ca_certs" in self.ssl_options:
- self.context.load_verify_locations(self.ssl_options["ca_certs"])
- if "cert_reqs" in self.ssl_options:
- self.context.set_verify(
- self.ssl_options["cert_reqs"],
- callback=self.verify_callback
- )
- self.context.set_info_callback(self.info_callback)
-
- def verify_callback(self, connection, x509, errnum, errdepth, ok):
- return ok
-
- def info_callback(self, connection, where, ret):
- if where & SSL.SSL_CB_HANDSHAKE_DONE:
- if self.check_hostname and self.endpoint.address != connection.get_peer_certificate().get_subject().commonName:
- transport = connection.get_app_data()
- transport.failVerification(Failure(ConnectionException("Hostname verification failed", self.endpoint)))
-
- def clientConnectionForTLS(self, tlsProtocol):
- connection = SSL.Connection(self.context, None)
- connection.set_app_data(tlsProtocol)
- if self.ssl_options and "server_hostname" in self.ssl_options:
- connection.set_tlsext_host_name(self.ssl_options['server_hostname'].encode('ascii'))
- return connection
-
-@deprecated(version="3.30.0", reason="The Twisted event loop is deprecated and will be removed in 3.31.0. See CASSPYTHON-12.")
-class TwistedConnection(Connection):
- """
- An implementation of :class:`.Connection` that utilizes the
- Twisted event loop.
- """
-
- _loop = None
-
- @classmethod
- def initialize_reactor(cls):
- if not cls._loop:
- cls._loop = TwistedLoop()
-
- @classmethod
- def create_timer(cls, timeout, callback):
- timer = Timer(timeout, callback)
- cls._loop.add_timer(timer)
- return timer
-
- def __init__(self, *args, **kwargs):
- """
- Initialization method.
-
- Note that we can't call reactor methods directly here because
- it's not thread-safe, so we schedule the reactor/connection
- stuff to be run from the event loop thread when it gets the
- chance.
- """
- Connection.__init__(self, *args, **kwargs)
-
- self.is_closed = True
- self.connector = None
- self.transport = None
-
- reactor.callFromThread(self.add_connection)
- self._loop.maybe_start()
-
- def _check_pyopenssl(self):
- if self.ssl_context or self.ssl_options:
- if not _HAS_SSL:
- raise ImportError(
- str(import_exception) +
- ', pyOpenSSL must be installed to enable SSL support with the Twisted event loop'
- )
-
- def add_connection(self):
- """
- Convenience function to connect and store the resulting
- connector.
- """
- host, port = self.endpoint.resolve()
- if self.ssl_context or self.ssl_options:
- # Can't use optionsForClientTLS here because it *forces* hostname verification.
- # Cool they enforce strong security, but we have to be able to turn it off
- self._check_pyopenssl()
-
- ssl_connection_creator = _SSLCreator(
- self.endpoint,
- self.ssl_context if self.ssl_context else None,
- self.ssl_options,
- self._check_hostname,
- self.connect_timeout,
- )
-
- endpoint = SSL4ClientEndpoint(
- reactor,
- host,
- port,
- sslContextFactory=ssl_connection_creator,
- timeout=self.connect_timeout,
- )
- else:
- endpoint = TCP4ClientEndpoint(
- reactor,
- host,
- port,
- timeout=self.connect_timeout
- )
- connectProtocol(endpoint, TwistedConnectionProtocol(self))
-
- def client_connection_made(self, transport):
- """
- Called by twisted protocol when a connection attempt has
- succeeded.
- """
- with self.lock:
- self.is_closed = False
- self.transport = transport
- self._send_options_message()
-
- def close(self):
- """
- Disconnect and error-out all requests.
- """
- with self.lock:
- if self.is_closed:
- return
- self.is_closed = True
-
- log.debug("Closing connection (%s) to %s", id(self), self.endpoint)
- reactor.callFromThread(self.transport.connector.disconnect)
- log.debug("Closed socket to %s", self.endpoint)
-
- if not self.is_defunct:
- self.error_all_requests(
- ConnectionShutdown("Connection to %s was closed" % self.endpoint))
- # don't leave in-progress operations hanging
- self.connected_event.set()
-
- def handle_read(self):
- """
- Process the incoming data buffer.
- """
- self.process_io_buffer()
-
- def push(self, data):
- """
- This function is called when outgoing data should be queued
- for sending.
-
- Note that we can't call transport.write() directly because
- it is not thread-safe, so we schedule it to run from within
- the event loop when it gets the chance.
- """
- reactor.callFromThread(self.transport.write, data)
diff --git a/docs/api/cassandra/io/eventletreactor.rst b/docs/api/cassandra/io/eventletreactor.rst
deleted file mode 100644
index 1ba742c7e9..0000000000
--- a/docs/api/cassandra/io/eventletreactor.rst
+++ /dev/null
@@ -1,7 +0,0 @@
-``cassandra.io.eventletreactor`` - ``eventlet``-compatible Connection
-=====================================================================
-
-.. module:: cassandra.io.eventletreactor
-
-.. autoclass:: EventletConnection
- :members:
diff --git a/docs/api/cassandra/io/geventreactor.rst b/docs/api/cassandra/io/geventreactor.rst
deleted file mode 100644
index 603affe140..0000000000
--- a/docs/api/cassandra/io/geventreactor.rst
+++ /dev/null
@@ -1,7 +0,0 @@
-``cassandra.io.geventreactor`` - ``gevent``-compatible Event Loop
-=================================================================
-
-.. module:: cassandra.io.geventreactor
-
-.. autoclass:: GeventConnection
- :members:
diff --git a/docs/api/cassandra/io/twistedreactor.rst b/docs/api/cassandra/io/twistedreactor.rst
deleted file mode 100644
index 24e93bd432..0000000000
--- a/docs/api/cassandra/io/twistedreactor.rst
+++ /dev/null
@@ -1,9 +0,0 @@
-``cassandra.io.twistedreactor`` - Twisted Event Loop
-====================================================
-
-.. module:: cassandra.io.twistedreactor
-
-.. class:: TwistedConnection
-
- An implementation of :class:`~cassandra.io.connection.Connection` that uses
- Twisted's reactor as its event loop.
diff --git a/docs/api/index.rst b/docs/api/index.rst
index 9e778d508c..c98cfd6b28 100644
--- a/docs/api/index.rst
+++ b/docs/api/index.rst
@@ -24,10 +24,7 @@ Core Driver
cassandra/timestamps
cassandra/io/asyncioreactor
cassandra/io/asyncorereactor
- cassandra/io/eventletreactor
cassandra/io/libevreactor
- cassandra/io/geventreactor
- cassandra/io/twistedreactor
.. _om_api:
diff --git a/docs/cloud.rst b/docs/cloud.rst
index 3230720ec9..46b416e53f 100644
--- a/docs/cloud.rst
+++ b/docs/cloud.rst
@@ -67,10 +67,6 @@ In most circumstances, the client code for interacting with an Astra cluster wil
Limitations
===========
-Event loops
-^^^^^^^^^^^
-Evenlet isn't yet supported for python 3.7+ due to an `issue in Eventlet