IGNITE-13967 Optimize allocations during parsing

Author: ivandasch

pyignite/api/affinity.py

@@ -55,11 +55,11 @@
 partition_mapping = StructArray([
     ('is_applicable', Bool),
 
-    ('cache_mapping', Conditional(lambda ctx: ctx['is_applicable'] == b'\x01',
+    ('cache_mapping', Conditional(lambda stream, ctx: stream.mem_view(*ctx['is_applicable']) == b'\x01',
                                   lambda ctx: ctx['is_applicable'] is True,
                                   cache_mapping, empty_cache_mapping)),
 
-    ('node_mapping', Conditional(lambda ctx: ctx['is_applicable'] == b'\x01',
+    ('node_mapping', Conditional(lambda stream, ctx: stream.mem_view(*ctx['is_applicable']) == b'\x01',
                                  lambda ctx: ctx['is_applicable'] is True,
                                  node_mapping, empty_node_mapping)),
 ])

pyignite/api/binary.py

@@ -57,31 +57,35 @@ def get_binary_type(conn: 'Connection', binary_type: Union[str, int], query_id=N
                                     following=[('type_exists', Bool)])
 
     with BinaryStream(conn.recv(), conn) as stream:
-        response_head_type, recv_buffer = response_head_struct.parse(stream)
-        response_head = response_head_type.from_buffer_copy(recv_buffer)
+        response_head_type, response_positions = response_head_struct.parse(stream)
+        response_head = response_head_type.from_buffer_copy(stream.mem_view(*response_positions))
+        init_pos, total_len = response_positions
+
         response_parts = []
         if response_head.type_exists:
-            resp_body_type, resp_body_buffer = body_struct.parse(stream)
+            resp_body_type, resp_body_positions = body_struct.parse(stream)
             response_parts.append(('body', resp_body_type))
-            resp_body = resp_body_type.from_buffer_copy(resp_body_buffer)
-            recv_buffer += resp_body_buffer
+            resp_body = resp_body_type.from_buffer_copy(stream.mem_view(*resp_body_positions))
+            total_len += resp_body_positions[1]
             if resp_body.is_enum:
-                resp_enum, resp_enum_buffer = enum_struct.parse(stream)
+                resp_enum, resp_enum_positions = enum_struct.parse(stream)
                 response_parts.append(('enums', resp_enum))
-                recv_buffer += resp_enum_buffer
-            resp_schema_type, resp_schema_buffer = schema_struct.parse(stream)
+                total_len += resp_enum_positions[1]
+
+            resp_schema_type, resp_schema_positions = schema_struct.parse(stream)
             response_parts.append(('schema', resp_schema_type))
-            recv_buffer += resp_schema_buffer
-
-    response_class = type(
-        'GetBinaryTypeResponse',
-        (response_head_type,),
-        {
-            '_pack_': 1,
-            '_fields_': response_parts,
-        }
-    )
-    response = response_class.from_buffer_copy(recv_buffer)
+            total_len += resp_schema_positions[1]
+
+        response_class = type(
+            'GetBinaryTypeResponse',
+            (response_head_type,),
+            {
+                '_pack_': 1,
+                '_fields_': response_parts,
+            }
+        )
+        response = response_class.from_buffer_copy(stream.mem_view(init_pos, total_len))
+
     result = APIResult(response)
     if result.status != 0:
         return result

pyignite/binary.py

@@ -169,6 +169,7 @@ def _from_python(self, stream):
             stream.seek(initial_pos + header.schema_offset)
             stream.write(schema)
 
+            self._buffer = bytes(stream.mem_view(initial_pos, stream.tell() - initial_pos))
             self._hashcode = header.hash_code
 
         def _setattr(self, attr_name: str, attr_value: Any):

pyignite/connection/__init__.py

@@ -205,9 +205,9 @@ def read_response(self) -> Union[dict, OrderedDict]:
             ('length', Int),
             ('op_code', Byte),
         ])
-        with BinaryStream(self.recv(), self) as response_stream:
-            start_class, start_buffer = response_start.parse(response_stream)
-            start = start_class.from_buffer_copy(start_buffer)
+        with BinaryStream(self.recv(), self) as stream:
+            start_class, start_positions = response_start.parse(stream)
+            start = start_class.from_buffer_copy(stream.mem_view(*start_positions))
             data = response_start.to_python(start)
             response_end = None
             if data['op_code'] == 0:
@@ -222,8 +222,8 @@ def read_response(self) -> Union[dict, OrderedDict]:
                     ('node_uuid', UUIDObject),
                 ])
             if response_end:
-                end_class, end_buffer = response_end.parse(response_stream)
-                end = end_class.from_buffer_copy(end_buffer)
+                end_class, end_positions = response_end.parse(stream)
+                end = end_class.from_buffer_copy(stream.mem_view(*end_positions))
                 data.update(response_end.to_python(end))
             return data
 

pyignite/datatypes/cache_properties.py

@@ -93,10 +93,12 @@ def build_header(cls):
 
     @classmethod
     def parse(cls, stream):
+        init_pos = stream.tell()
+
         header_class = cls.build_header()
-        buf = stream.read(ctypes.sizeof(header_class))
-        data_class, data_buf = cls.prop_data_class.parse(stream)
-        buf += data_buf
+        header_len = ctypes.sizeof(header_class)
+        data_class, data_buf = cls.prop_data_class.parse(stream.mem_view(init_pos, header_len))
+
         prop_class = type(
             cls.__name__,
             (header_class,),
@@ -107,7 +109,9 @@ def parse(cls, stream):
                 ],
             }
         )
-        return prop_class, buf
+
+        stream.seek(init_pos + ctypes.sizeof(prop_class))
+        return prop_class, (init_pos, stream.tell() - init_pos)
 
     @classmethod
     def to_python(cls, ctype_object, *args, **kwargs):

pyignite/datatypes/complex.py

@@ -15,7 +15,7 @@
 
 from collections import OrderedDict
 import ctypes
-import inspect
+from io import SEEK_CUR
 from typing import Iterable, Dict
 
 from pyignite.constants import *
@@ -71,19 +71,20 @@ def build_header(cls):
 
     @classmethod
     def parse(cls, stream):
-        buffer = stream.read(ctypes.sizeof(ctypes.c_byte))
+        init_pos, type_len = stream.tell(), ctypes.sizeof(ctypes.c_byte)
 
+        buffer = stream.read(type_len)
         if buffer == TC_NULL:
-            return Null.build_c_type(), buffer
+            return Null.build_c_type(), (init_pos, type_len)
 
         header_class = cls.build_header()
-        buffer += stream.read(ctypes.sizeof(header_class) - len(buffer))
-        header = header_class.from_buffer_copy(buffer)
-        fields = []
+        header_len = ctypes.sizeof(header_class)
+        header = header_class.from_buffer_copy(stream.mem_view(init_pos, header_len))
+        stream.seek(init_pos + header_len)
 
+        fields = []
         for i in range(header.length):
-            c_type, buffer_fragment = AnyDataObject.parse(stream)
-            buffer += buffer_fragment
+            c_type, _ = AnyDataObject.parse(stream)
             fields.append(('element_{}'.format(i), c_type))
 
         final_class = type(
@@ -94,7 +95,9 @@ def parse(cls, stream):
                 '_fields_': fields,
             }
         )
-        return final_class, buffer
+
+        stream.seek(init_pos + ctypes.sizeof(final_class))
+        return final_class, (init_pos, stream.tell() - init_pos)
 
     @classmethod
     def to_python(cls, ctype_object, *args, **kwargs):
@@ -164,14 +167,16 @@ def build_header(cls):
 
     @classmethod
     def parse(cls, stream):
-        buffer = stream.read(ctypes.sizeof(ctypes.c_byte))
+        init_pos, type_len = stream.tell(), ctypes.sizeof(ctypes.c_byte)
 
+        buffer = stream.read(type_len)
         if buffer == TC_NULL:
-            return Null.build_c_type(), buffer
+            return Null.build_c_type(), (init_pos, type_len)
 
         header_class = cls.build_header()
-        buffer += stream.read(ctypes.sizeof(header_class) - len(buffer))
-        header = header_class.from_buffer_copy(buffer)
+        header_len = ctypes.sizeof(header_class)
+        header = header_class.from_buffer_copy(stream.mem_view(init_pos, header_len))
+        stream.seek(init_pos + header_len)
 
         final_class = type(
             cls.__name__,
@@ -184,10 +189,9 @@ def parse(cls, stream):
                 ],
             }
         )
-        buffer += stream.read(
-            ctypes.sizeof(final_class) - ctypes.sizeof(header_class)
-        )
-        return final_class, buffer
+
+        stream.seek(init_pos + ctypes.sizeof(final_class))
+        return final_class, (init_pos, stream.tell() - init_pos)
 
     @classmethod
     def to_python(cls, ctype_object, *args, **kwargs):
@@ -262,19 +266,21 @@ def build_header(cls):
 
     @classmethod
     def parse(cls, stream):
-        buffer = stream.read(ctypes.sizeof(ctypes.c_byte))
+        init_pos, type_len = stream.tell(), ctypes.sizeof(ctypes.c_byte)
 
-        if buffer == TC_NULL:
-            return Null.build_c_type(), buffer
+        type_ = stream.mem_view(init_pos, type_len)
+        if type_ == TC_NULL:
+            stream.seek(type_len, SEEK_CUR)
+            return Null.build_c_type(), (init_pos, type_len)
 
         header_class = cls.build_header()
-        buffer += stream.read(ctypes.sizeof(header_class) - len(buffer))
-        header = header_class.from_buffer_copy(buffer)
-        fields = []
+        header_len = ctypes.sizeof(header_class)
+        header = header_class.from_buffer_copy(stream.mem_view(init_pos, header_len))
+        stream.seek(init_pos + header_len)
 
+        fields = []
         for i in range(header.length):
-            c_type, buffer_fragment = AnyDataObject.parse(stream)
-            buffer += buffer_fragment
+            c_type, _ = AnyDataObject.parse(stream)
             fields.append(('element_{}'.format(i), c_type))
 
         final_class = type(
@@ -285,7 +291,7 @@ def parse(cls, stream):
                 '_fields_': fields,
             }
         )
-        return final_class, buffer
+        return final_class, (init_pos, stream.tell() - init_pos)
 
     @classmethod
     def to_python(cls, ctype_object, *args, **kwargs):
@@ -360,19 +366,20 @@ def build_header(cls):
 
     @classmethod
     def parse(cls, stream):
-        buffer = stream.read(ctypes.sizeof(ctypes.c_byte))
+        init_pos, type_len = stream.tell(), ctypes.sizeof(ctypes.c_byte)
 
+        buffer = stream.read(type_len)
         if buffer == TC_NULL:
-            return Null.build_c_type(), buffer
+            return Null.build_c_type(), (init_pos, type_len)
 
         header_class = cls.build_header()
-        buffer += stream.read(ctypes.sizeof(header_class) - len(buffer))
-        header = header_class.from_buffer_copy(buffer)
-        fields = []
+        header_len = ctypes.sizeof(header_class)
+        header = header_class.from_buffer_copy(stream.mem_view(init_pos, header_len))
+        stream.seek(init_pos + header_len)
 
+        fields = []
         for i in range(header.length << 1):
-            c_type, buffer_fragment = AnyDataObject.parse(stream)
-            buffer += buffer_fragment
+            c_type, _ = AnyDataObject.parse(stream)
             fields.append(('element_{}'.format(i), c_type))
 
         final_class = type(
@@ -383,7 +390,7 @@ def parse(cls, stream):
                 '_fields_': fields,
             }
         )
-        return final_class, buffer
+        return final_class, (init_pos, stream.tell() - init_pos)
 
     @classmethod
     def to_python(cls, ctype_object, *args, **kwargs):
@@ -541,27 +548,30 @@ def schema_type(cls, flags: int):
     @classmethod
     def parse(cls, stream):
         from pyignite.datatypes import Struct
-        buffer = stream.read(ctypes.sizeof(ctypes.c_byte))
 
-        if buffer == TC_NULL:
-            return Null.build_c_type(), buffer
+        init_pos, type_len = stream.tell(), ctypes.sizeof(ctypes.c_byte)
+
+        type_ = stream.mem_view(init_pos, type_len)
+        if type_ == TC_NULL:
+            stream.seek(type_len, SEEK_CUR)
+            return Null.build_c_type(), (init_pos, type_len)
 
         header_class = cls.build_header()
-        buffer += stream.read(ctypes.sizeof(header_class) - len(buffer))
-        header = header_class.from_buffer_copy(buffer)
+        header_len = ctypes.sizeof(header_class)
+        header = header_class.from_buffer_copy(stream.mem_view(init_pos, header_len))
+        stream.seek(init_pos + header_len)
 
         # ignore full schema, always retrieve fields' types and order
         # from complex types registry
         data_class = stream.get_dataclass(header)
         fields = data_class.schema.items()
         object_fields_struct = Struct(fields)
-        object_fields, object_fields_buffer = object_fields_struct.parse(stream)
-        buffer += object_fields_buffer
+        object_fields, _ = object_fields_struct.parse(stream)
         final_class_fields = [('object_fields', object_fields)]
 
         if header.flags & cls.HAS_SCHEMA:
             schema = cls.schema_type(header.flags) * len(fields)
-            buffer += stream.read(ctypes.sizeof(schema))
+            stream.seek(ctypes.sizeof(schema), SEEK_CUR)
             final_class_fields.append(('schema', schema))
 
         final_class = type(
@@ -574,7 +584,7 @@ def parse(cls, stream):
         )
         # register schema encoding approach
         stream.compact_footer = bool(header.flags & cls.COMPACT_FOOTER)
-        return final_class, buffer
+        return final_class, (init_pos, stream.tell() - init_pos)
 
     @classmethod
     def to_python(cls, ctype_object, client: 'Client' = None, *args, **kwargs):
@@ -610,4 +620,7 @@ def from_python(cls, stream, value):
             return
 
         stream.register_binary_type(value.__class__)
-        value._from_python(stream)
+        if getattr(value, '_buffer', None):
+            stream.write(value._buffer)
+        else:
+            value._from_python(stream)