WIP

Author: mykaul

benchmarks/callback_full_pipeline.py

@@ -49,10 +49,7 @@ def insert_next(self, previous_result=sentinel):
     def run(self):
         self.start_profile()
 
-        if self.protocol_version >= 3:
-            concurrency = 1000
-        else:
-            concurrency = 100
+        concurrency = 1000
 
         for _ in range(min(concurrency, self.num_queries)):
             self.insert_next()

cassandra/__init__.py

@@ -135,16 +135,6 @@ class ProtocolVersion(object):
     """
     Defines native protocol versions supported by this driver.
     """
-    V1 = 1
-    """
-    v1, supported in Cassandra 1.2-->2.2
-    """
-
-    V2 = 2
-    """
-    v2, supported in Cassandra 2.0-->2.2;
-    added support for lightweight transactions, batch operations, and automatic query paging.
-    """
 
     V3 = 3
     """
@@ -180,7 +170,7 @@ class ProtocolVersion(object):
     DSE private protocol v2, supported in DSE 6.0+
     """
 
-    SUPPORTED_VERSIONS = (DSE_V2, DSE_V1, V6, V5, V4, V3, V2, V1)
+    SUPPORTED_VERSIONS = (DSE_V2, DSE_V1, V6, V5, V4, V3)
     """
     A tuple of all supported protocol versions
     """

cassandra/cluster.py

@@ -1452,18 +1452,6 @@ def __init__(self,
 
         self._user_types = defaultdict(dict)
 
-        self._min_requests_per_connection = {
-            HostDistance.LOCAL_RACK: DEFAULT_MIN_REQUESTS,
-            HostDistance.LOCAL: DEFAULT_MIN_REQUESTS,
-            HostDistance.REMOTE: DEFAULT_MIN_REQUESTS
-        }
-
-        self._max_requests_per_connection = {
-            HostDistance.LOCAL_RACK: DEFAULT_MAX_REQUESTS,
-            HostDistance.LOCAL: DEFAULT_MAX_REQUESTS,
-            HostDistance.REMOTE: DEFAULT_MAX_REQUESTS
-        }
-
         self._core_connections_per_host = {
             HostDistance.LOCAL_RACK: DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST,
             HostDistance.LOCAL: DEFAULT_MIN_CONNECTIONS_PER_LOCAL_HOST,
@@ -1666,48 +1654,6 @@ def add_execution_profile(self, name, profile, pool_wait_timeout=5):
         if not_done:
             raise OperationTimedOut("Failed to create all new connection pools in the %ss timeout.")
 
-    def get_min_requests_per_connection(self, host_distance):
-        return self._min_requests_per_connection[host_distance]
-
-    def set_min_requests_per_connection(self, host_distance, min_requests):
-        """
-        Sets a threshold for concurrent requests per connection, below which
-        connections will be considered for disposal (down to core connections;
-        see :meth:`~Cluster.set_core_connections_per_host`).
-
-        Pertains to connection pool management in protocol versions {1,2}.
-        """
-        if self.protocol_version >= 3:
-            raise UnsupportedOperation(
-                "Cluster.set_min_requests_per_connection() only has an effect "
-                "when using protocol_version 1 or 2.")
-        if min_requests < 0 or min_requests > 126 or \
-           min_requests >= self._max_requests_per_connection[host_distance]:
-            raise ValueError("min_requests must be 0-126 and less than the max_requests for this host_distance (%d)" %
-                             (self._min_requests_per_connection[host_distance],))
-        self._min_requests_per_connection[host_distance] = min_requests
-
-    def get_max_requests_per_connection(self, host_distance):
-        return self._max_requests_per_connection[host_distance]
-
-    def set_max_requests_per_connection(self, host_distance, max_requests):
-        """
-        Sets a threshold for concurrent requests per connection, above which new
-        connections will be created to a host (up to max connections;
-        see :meth:`~Cluster.set_max_connections_per_host`).
-
-        Pertains to connection pool management in protocol versions {1,2}.
-        """
-        if self.protocol_version >= 3:
-            raise UnsupportedOperation(
-                "Cluster.set_max_requests_per_connection() only has an effect "
-                "when using protocol_version 1 or 2.")
-        if max_requests < 1 or max_requests > 127 or \
-           max_requests <= self._min_requests_per_connection[host_distance]:
-            raise ValueError("max_requests must be 1-127 and greater than the min_requests for this host_distance (%d)" %
-                             (self._min_requests_per_connection[host_distance],))
-        self._max_requests_per_connection[host_distance] = max_requests
-
     def get_core_connections_per_host(self, host_distance):
         """
         Gets the minimum number of connections per Session that will be opened

cassandra/connection.py

@@ -123,7 +123,6 @@ def decompress(byts):
 DEFAULT_LOCAL_PORT_LOW = 49152
 DEFAULT_LOCAL_PORT_HIGH = 65535
 
-frame_header_v1_v2 = struct.Struct('>BbBi')
 frame_header_v3 = struct.Struct('>BhBi')
 
 
@@ -817,17 +816,12 @@ def __init__(self, host='127.0.0.1', port=9042, authenticator=None,
         if not self.ssl_context and self.ssl_options:
             self.ssl_context = self._build_ssl_context_from_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.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
 
         self.lock = RLock()
         self.connected_event = Event()
@@ -1205,11 +1199,10 @@ def _read_frame_header(self):
             version = buf[0] & PROTOCOL_VERSION_MASK
             if version not in ProtocolVersion.SUPPORTED_VERSIONS:
                 raise ProtocolError("This version of the driver does not support protocol version %d" % version)
-            frame_header = frame_header_v3 if version >= 3 else frame_header_v1_v2
             # this frame header struct is everything after the version byte
-            header_size = frame_header.size + 1
+            header_size = frame_header_v3.size + 1
             if pos >= header_size:
-                flags, stream, op, body_len = frame_header.unpack_from(buf, 1)
+                flags, stream, op, body_len = frame_header_v3.unpack_from(buf, 1)
                 if body_len < 0:
                     raise ProtocolError("Received negative body length: %r" % body_len)
                 self._current_frame = _Frame(version, flags, stream, op, header_size, body_len + header_size)

cassandra/cqltypes.py

@@ -812,18 +812,13 @@ class _SimpleParameterizedType(_ParameterizedType):
     @classmethod
     def deserialize_safe(cls, byts, protocol_version):
         subtype, = cls.subtypes
-        if protocol_version >= 3:
-            unpack = int32_unpack
-            length = 4
-        else:
-            unpack = uint16_unpack
-            length = 2
-        numelements = unpack(byts[:length])
+        length = 4
+        numelements = int32_unpack(byts[:length])
         p = length
         result = []
         inner_proto = max(3, protocol_version)
         for _ in range(numelements):
-            itemlen = unpack(byts[p:p + length])
+            itemlen = int32_unpack(byts[p:p + length])
             p += length
             if itemlen < 0:
                 result.append(None)
@@ -839,16 +834,15 @@ def serialize_safe(cls, items, protocol_version):
             raise TypeError("Received a string for a type that expects a sequence")
 
         subtype, = cls.subtypes
-        pack = int32_pack if protocol_version >= 3 else uint16_pack
         buf = io.BytesIO()
-        buf.write(pack(len(items)))
+        buf.write(int32_pack(len(items)))
         inner_proto = max(3, protocol_version)
         for item in items:
             if item is None:
-                buf.write(pack(-1))
+                buf.write(int32_pack(-1))
             else:
                 itembytes = subtype.to_binary(item, inner_proto)
-                buf.write(pack(len(itembytes)))
+                buf.write(int32_pack(len(itembytes)))
                 buf.write(itembytes)
         return buf.getvalue()
 
@@ -872,18 +866,13 @@ class MapType(_ParameterizedType):
     @classmethod
     def deserialize_safe(cls, byts, protocol_version):
         key_type, value_type = cls.subtypes
-        if protocol_version >= 3:
-            unpack = int32_unpack
-            length = 4
-        else:
-            unpack = uint16_unpack
-            length = 2
-        numelements = unpack(byts[:length])
+        length = 4
+        numelements = int32_unpack(byts[:length])
         p = length
         themap = util.OrderedMapSerializedKey(key_type, protocol_version)
         inner_proto = max(3, protocol_version)
         for _ in range(numelements):
-            key_len = unpack(byts[p:p + length])
+            key_len = int32_unpack(byts[p:p + length])
             p += length
             if key_len < 0:
                 keybytes = None
@@ -893,7 +882,7 @@ def deserialize_safe(cls, byts, protocol_version):
                 p += key_len
                 key = key_type.from_binary(keybytes, inner_proto)
 
-            val_len = unpack(byts[p:p + length])
+            val_len = int32_unpack(byts[p:p + length])
             p += length
             if val_len < 0:
                 val = None
@@ -908,9 +897,8 @@ def deserialize_safe(cls, byts, protocol_version):
     @classmethod
     def serialize_safe(cls, themap, protocol_version):
         key_type, value_type = cls.subtypes
-        pack = int32_pack if protocol_version >= 3 else uint16_pack
         buf = io.BytesIO()
-        buf.write(pack(len(themap)))
+        buf.write(int32_pack(len(themap)))
         try:
             items = themap.items()
         except AttributeError:
@@ -919,16 +907,16 @@ def serialize_safe(cls, themap, protocol_version):
         for key, val in items:
             if key is not None:
                 keybytes = key_type.to_binary(key, inner_proto)
-                buf.write(pack(len(keybytes)))
+                buf.write(int32_pack(len(keybytes)))
                 buf.write(keybytes)
             else:
-                buf.write(pack(-1))
+                buf.write(int32_pack(-1))
             if val is not None:
                 valbytes = value_type.to_binary(val, inner_proto)
-                buf.write(pack(len(valbytes)))
+                buf.write(int32_pack(len(valbytes)))
                 buf.write(valbytes)
             else:
-                buf.write(pack(-1))
+                buf.write(int32_pack(-1))
         return buf.getvalue()
 
 

cassandra/metadata.py

@@ -153,12 +153,7 @@ def refresh(self, connection, timeout, target_type=None, change_type=None, fetch
             meta = parse_method(self.keyspaces, **kwargs)
             if meta:
                 update_method = getattr(self, '_update_' + tt_lower)
-                if tt_lower == 'keyspace' and connection.protocol_version < 3:
-                    # we didn't have 'type' target in legacy protocol versions, so we need to query those too
-                    user_types = parser.get_types_map(self.keyspaces, **kwargs)
-                    self._update_keyspace(meta, user_types)
-                else:
-                    update_method(meta)
+                update_method(meta)
             else:
                 drop_method = getattr(self, '_drop_' + tt_lower)
                 drop_method(**kwargs)

cassandra/pool.py

@@ -990,7 +990,6 @@ def borrow_connection(self, timeout, routing_key=None):
             # note: it would be nice to push changes to these config settings
             # to pools instead of doing a new lookup on every
             # borrow_connection() call
-            max_reqs = self._session.cluster.get_max_requests_per_connection(self.host_distance)
             max_conns = self._session.cluster.get_max_connections_per_host(self.host_distance)
 
             least_busy = min(conns, key=lambda c: c.in_flight)
@@ -1012,10 +1011,9 @@ def borrow_connection(self, timeout, routing_key=None):
                 # wait_for_conn will increment in_flight on the conn
                 least_busy, request_id = self._wait_for_conn(timeout)
 
-            # if we have too many requests on this connection but we still
-            # have space to open a new connection against this host, go ahead
+            # if we still have space to open a new connection against this host, go ahead
             # and schedule the creation of a new connection
-            if least_busy.in_flight >= max_reqs and len(self._connections) < max_conns:
+            if len(self._connections) < max_conns:
                 self._maybe_spawn_new_connection()
 
             return least_busy, request_id
@@ -1144,11 +1142,10 @@ def return_connection(self, connection, stream_was_orphaned=False):
                 return
 
             core_conns = self._session.cluster.get_core_connections_per_host(self.host_distance)
-            min_reqs = self._session.cluster.get_min_requests_per_connection(self.host_distance)
             # we can use in_flight here without holding the connection lock
             # because the fact that in_flight dipped below the min at some
             # point is enough to start the trashing procedure
-            if len(self._connections) > core_conns and in_flight <= min_reqs and \
+            if len(self._connections) > core_conns and \
                     time.time() >= self._next_trash_allowed_at:
                 self._maybe_trash_connection(connection)
             else: