# Copyright 2014 Symantec Corporation # 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. # 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 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)) ) 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])