Merge pull request #299 from YorikSar/conn-speedup

Protocol parsing optimizations

Author: zzzeek

pymysql/connections.py

@@ -39,7 +39,8 @@
 from .cursors import Cursor
 from .constants import CLIENT, COMMAND, FIELD_TYPE, SERVER_STATUS
 from .util import byte2int, int2byte
-from .converters import escape_item, encoders, decoders, escape_string
+from .converters import (
+    escape_item, encoders, decoders, escape_string, through)
 from .err import (
     raise_mysql_exception, Warning, Error,
     InterfaceError, DataError, DatabaseError, OperationalError,
@@ -102,10 +103,6 @@ def _makefile(sock, mode):
 UNSIGNED_SHORT_COLUMN = 252
 UNSIGNED_INT24_COLUMN = 253
 UNSIGNED_INT64_COLUMN = 254
-UNSIGNED_CHAR_LENGTH = 1
-UNSIGNED_SHORT_LENGTH = 2
-UNSIGNED_INT24_LENGTH = 3
-UNSIGNED_INT64_LENGTH = 8
 
 DEFAULT_CHARSET = 'latin1'
 
@@ -216,18 +213,6 @@ def _hash_password_323(password):
 def pack_int24(n):
     return struct.pack('<I', n)[:3]
 
-def unpack_uint16(n):
-    return struct.unpack('<H', n[0:2])[0]
-
-def unpack_int24(n):
-    return struct.unpack('<I', n + b'\0')[0]
-
-def unpack_int32(n):
-    return struct.unpack('<I', n)[0]
-
-def unpack_int64(n):
-    return struct.unpack('<Q', n)[0]
-
 
 class MysqlPacket(object):
     """Representation of a MySQL response packet.
@@ -291,23 +276,54 @@ def get_bytes(self, position, length=1):
         """
         return self._data[position:(position+length)]
 
+    if PY2:
+        def read_uint8(self):
+            result = ord(self._data[self._position])
+            self._position += 1
+            return result
+    else:
+        def read_uint8(self):
+            result = self._data[self._position]
+            self._position += 1
+            return result
+
+    def read_uint16(self):
+        result = struct.unpack_from('<H', self._data, self._position)[0]
+        self._position += 2
+        return result
+
+    def read_uint24(self):
+        low, high = struct.unpack_from('<HB', self._data, self._position)
+        self._position += 3
+        return low + (high << 16)
+
+    def read_uint32(self):
+        result = struct.unpack_from('<I', self._data, self._position)[0]
+        self._position += 4
+        return result
+
+    def read_uint64(self):
+        result = struct.unpack_from('<Q', self._data, self._position)[0]
+        self._position += 8
+        return result
+
     def read_length_encoded_integer(self):
         """Read a 'Length Coded Binary' number from the data buffer.
 
         Length coded numbers can be anywhere from 1 to 9 bytes depending
         on the value of the first byte.
         """
-        c = ord(self.read(1))
+        c = self.read_uint8()
         if c == NULL_COLUMN:
             return None
         if c < UNSIGNED_CHAR_COLUMN:
             return c
         elif c == UNSIGNED_SHORT_COLUMN:
-            return unpack_uint16(self.read(UNSIGNED_SHORT_LENGTH))
+            return self.read_uint16()
         elif c == UNSIGNED_INT24_COLUMN:
-            return unpack_int24(self.read(UNSIGNED_INT24_LENGTH))
+            return self.read_uint24()
         elif c == UNSIGNED_INT64_COLUMN:
-            return unpack_int64(self.read(UNSIGNED_INT64_LENGTH))
+            return self.read_uint64()
 
     def read_length_coded_string(self):
         """Read a 'Length Coded String' from the data buffer.
@@ -321,6 +337,12 @@ def read_length_coded_string(self):
             return None
         return self.read(length)
 
+    def read_struct(self, fmt):
+        s = struct.Struct(fmt)
+        result = s.unpack_from(self._data, self._position)
+        self._position += s.size
+        return result
+
     def is_ok_packet(self):
         return self._data[0:1] == b'\0'
 
@@ -344,7 +366,7 @@ def check_error(self):
         if self.is_error_packet():
             self.rewind()
             self.advance(1)  # field_count == error (we already know that)
-            errno = unpack_uint16(self.read(2))
+            errno = self.read_uint16()
             if DEBUG: print("errno =", errno)
             raise_mysql_exception(self._data)
 
@@ -374,13 +396,8 @@ def __parse_field_descriptor(self, encoding):
         self.org_table = self.read_length_coded_string().decode(encoding)
         self.name = self.read_length_coded_string().decode(encoding)
         self.org_name = self.read_length_coded_string().decode(encoding)
-        self.advance(1)  # non-null filler
-        self.charsetnr = struct.unpack('<H', self.read(2))[0]
-        self.length = struct.unpack('<I', self.read(4))[0]
-        self.type_code = byte2int(self.read(1))
-        self.flags = struct.unpack('<H', self.read(2))[0]
-        self.scale = byte2int(self.read(1))  # "decimals"
-        self.advance(2)  # filler (always 0x00)
+        self.charsetnr, self.length, self.type_code, self.flags, self.scale = (
+            self.read_struct('<xHIBHBxx'))
         # 'default' is a length coded binary and is still in the buffer?
         # not used for normal result sets...
 
@@ -424,8 +441,7 @@ def __init__(self, from_packet):
 
         self.affected_rows = self.packet.read_length_encoded_integer()
         self.insert_id = self.packet.read_length_encoded_integer()
-        self.server_status = struct.unpack('<H', self.packet.read(2))[0]
-        self.warning_count = struct.unpack('<H', self.packet.read(2))[0]
+        self.server_status, self.warning_count = self.read_struct('<HH')
         self.message = self.packet.read_all()
         self.has_next = self.server_status & SERVER_STATUS.SERVER_MORE_RESULTS_EXISTS
 
@@ -447,9 +463,7 @@ def __init__(self, from_packet):
                     self.__class__))
 
         self.packet = from_packet
-        from_packet.advance(1)
-        self.warning_count = struct.unpack('<h', from_packet.read(2))[0]
-        self.server_status = struct.unpack('<h', self.packet.read(2))[0]
+        self.warning_count, self.server_status = self.packet.read_struct('<xhh')
         if DEBUG: print("server_status=", self.server_status)
         self.has_next = self.server_status & SERVER_STATUS.SERVER_MORE_RESULTS_EXISTS
 
@@ -635,7 +649,7 @@ def close(self):
         ''' Send the quit message and close the socket '''
         if self.socket is None:
             raise Error("Already closed")
-        send_data = struct.pack('<i', 1) + int2byte(COMMAND.COM_QUIT)
+        send_data = struct.pack('<iB', 1, COMMAND.COM_QUIT)
         try:
             self._write_bytes(send_data)
         except Exception:
@@ -854,14 +868,9 @@ def _read_packet(self, packet_type=MysqlPacket):
         while True:
             packet_header = self._read_bytes(4)
             if DEBUG: dump_packet(packet_header)
-            packet_length_bin = packet_header[:3]
-
+            btrl, btrh, packet_number = struct.unpack('<HBB', packet_header)
+            bytes_to_read = btrl + (btrh << 16)
             #TODO: check sequence id
-            #  packet_number
-            byte2int(packet_header[3])
-
-            bin_length = packet_length_bin + b'\0'  # pad little-endian number
-            bytes_to_read = struct.unpack('<I', bin_length)[0]
             recv_data = self._read_bytes(bytes_to_read)
             if DEBUG: dump_packet(recv_data)
             buff += recv_data
@@ -930,7 +939,7 @@ def _execute_command(self, command, sql):
 
         chunk_size = min(MAX_PACKET_LEN, len(sql) + 1)  # +1 is for command
 
-        prelude = struct.pack('<i', chunk_size) + int2byte(command)
+        prelude = struct.pack('<iB', chunk_size, command)
         self._write_bytes(prelude + sql[:chunk_size-1])
         if DEBUG: dump_packet(prelude + sql)
 
@@ -962,8 +971,7 @@ def _request_authentication(self):
         if isinstance(self.user, text_type):
             self.user = self.user.encode(self.encoding)
 
-        data_init = (struct.pack('<i', self.client_flag) + struct.pack("<I", 1) +
-                     int2byte(charset_id) + int2byte(0)*23)
+        data_init = struct.pack('<iIB23s', self.client_flag, 1, charset_id, b'')
 
         next_packet = 1
 
@@ -1202,23 +1210,12 @@ def _read_rowdata_packet(self):
         self.rows = tuple(rows)
 
     def _read_row_from_packet(self, packet):
-        use_unicode = self.connection.use_unicode
         row = []
-        for field in self.fields:
+        for encoding, converter in self.converters:
             data = packet.read_length_coded_string()
             if data is not None:
-                field_type = field.type_code
-                if use_unicode:
-                    if field_type in TEXT_TYPES:
-                        charset = charset_by_id(field.charsetnr)
-                        if use_unicode and not charset.is_binary:
-                            # TEXTs with charset=binary means BINARY types.
-                            data = data.decode(charset.encoding)
-                    else:
-                        data = data.decode()
-
-                converter = self.connection.decoders.get(field_type)
-                if DEBUG: print("DEBUG: field={}, converter={}".format(field, converter))
+                if encoding is not None:
+                    data = data.decode(encoding)
                 if DEBUG: print("DEBUG: DATA = ", data)
                 if converter is not None:
                     data = converter(data)
@@ -1228,11 +1225,31 @@ def _read_row_from_packet(self, packet):
     def _get_descriptions(self):
         """Read a column descriptor packet for each column in the result."""
         self.fields = []
+        self.converters = []
+        use_unicode = self.connection.use_unicode
         description = []
         for i in range_type(self.field_count):
             field = self.connection._read_packet(FieldDescriptorPacket)
             self.fields.append(field)
             description.append(field.description())
+            field_type = field.type_code
+            if use_unicode:
+                if field_type in TEXT_TYPES:
+                    charset = charset_by_id(field.charsetnr)
+                    if charset.is_binary:
+                        # TEXTs with charset=binary means BINARY types.
+                        encoding = None
+                    else:
+                        encoding = charset.encoding
+                else:
+                    encoding = 'ascii'
+            else:
+                encoding = None
+            converter = self.connection.decoders.get(field_type)
+            if converter is through:
+                converter = None
+            if DEBUG: print("DEBUG: field={}, converter={}".format(field, converter))
+            self.converters.append((encoding, converter))
 
         eof_packet = self.connection._read_packet()
         assert eof_packet.is_eof_packet(), 'Protocol error, expecting EOF'

pymysql/converters.py

@@ -254,13 +254,19 @@ def convert_mysql_timestamp(timestamp):
 def convert_set(s):
     return set(s.split(","))
 
-def convert_bit(b):
-    #b = "\x00" * (8 - len(b)) + b # pad w/ zeroes
-    #return struct.unpack(">Q", b)[0]
-    #
-    # the snippet above is right, but MySQLdb doesn't process bits,
-    # so we shouldn't either
-    return b
+
+def through(x):
+    return x
+
+
+#def convert_bit(b):
+#    b = "\x00" * (8 - len(b)) + b # pad w/ zeroes
+#    return struct.unpack(">Q", b)[0]
+#    
+#     the snippet above is right, but MySQLdb doesn't process bits,
+#     so we shouldn't either
+convert_bit = through
+
 
 def convert_characters(connection, field, data):
     field_charset = charset_by_id(field.charsetnr).name
@@ -297,10 +303,6 @@ def convert_characters(connection, field, data):
     Decimal: str,
 }
 
-
-def through(x):
-    return x
-
 if not PY2 or JYTHON or IRONPYTHON:
     encoders[bytes] = escape_bytes