Improve performance of serialization.

(a) we reduce the number of times we have to copy the data into intermediate
    buffers.
(b) we allow things like ListOf(int) to use a fastpath and avoid having to
    varint encode everything (which does increase the size but is faster)

Author: braxtonmckee

typed_python/NullSerializationContext.hpp

@@ -37,6 +37,9 @@ class NullSerializationContext : public SerializationContext {
     }
 
 
+    virtual bool serializePodListsInline() const {
+        return true;
+    }
     virtual bool isCompressionEnabled() const {
         return false;
     }

typed_python/PythonSerializationContext.cpp

@@ -38,6 +38,14 @@ void PythonSerializationContext::setFlags() {
 
     mSerializeHashSequence = ((PyObject*)serializeHashSequence) == Py_True;
 
+    PyObjectStealer serializePodListsInline(PyObject_GetAttrString(mContextObj, "serializePodListsInline"));
+
+    if (!serializePodListsInline) {
+        throw PythonExceptionSet();
+    }
+
+    mSerializePodListsInline = ((PyObject*)serializePodListsInline) == Py_True;
+
     PyObjectStealer encodeLineInformationForCode(PyObject_GetAttrString(mContextObj, "encodeLineInformationForCode"));
 
     if (!encodeLineInformationForCode) {

typed_python/PythonSerializationContext.hpp

@@ -83,6 +83,7 @@ class PythonSerializationContext : public SerializationContext {
     PythonSerializationContext(PyObject* typeSetObj) :
             mContextObj(typeSetObj),
             mCompressionEnabled(false),
+            mSerializePodListsInline(false),
             mSerializeHashSequence(false)
     {
         setFlags();
@@ -98,6 +99,10 @@ class PythonSerializationContext : public SerializationContext {
         return mSuppressLineInfo;
     }
 
+    bool serializePodListsInline() const {
+        return mSerializePodListsInline;
+    }
+
     // should we serialize an integer in the order of the
     // hash sequence rather than the hash itself?
     bool shouldSerializeHashSequence() const {
@@ -189,6 +194,8 @@ class PythonSerializationContext : public SerializationContext {
 
     bool mCompressionEnabled;
 
+    bool mSerializePodListsInline;
+
     bool mSuppressLineInfo;
 
     bool mSerializeHashSequence;

typed_python/RegisterTypes.hpp

@@ -263,6 +263,13 @@ class RegisterType : public Type {
         buffer.writeRegisterType(fieldNumber, *(T*)self);
     }
 
+    template<class buf_t>
+    void serializeMultiConcrete(instance_ptr left, size_t count, size_t stride, buf_t& buffer, size_t fieldNumber) {
+        for (long k = 0; k < count; k++) {
+            buffer.writeRegisterType(fieldNumber, *(T*)(left + k * stride));
+        }
+    }
+
     size_t deepBytecountConcrete(instance_ptr instance, std::unordered_set<void*>& alreadyVisited, std::set<Slab*>* outSlabs) {
         return 0;
     }

typed_python/SerializationBuffer.cpp

@@ -24,8 +24,8 @@ Bytes SerializationBuffer::serializeSingleBoolToBytes(bool value) {
     return Bytes((const char*)existsValue, 2);
 }
 
-void SerializationBuffer::compress() {
-    if (m_last_compression_point == m_size) {
+void SerializationBufferBlock::compress() {
+    if (m_compressed) {
         return;
     }
 
@@ -37,7 +37,7 @@ void SerializationBuffer::compress() {
 
     //replace the data we have here with a block of 4 bytes of size of compressed data and
     //then the data stream
-    size_t bytesRequired = LZ4F_compressFrameBound(m_size - m_last_compression_point, &lz4Prefs);
+    size_t bytesRequired = LZ4F_compressFrameBound(m_size, &lz4Prefs);
 
     void* compressedBytes = malloc(bytesRequired);
 
@@ -49,26 +49,64 @@ void SerializationBuffer::compress() {
         compressedBytecount = LZ4F_compressFrame(
             compressedBytes,
             bytesRequired,
-            m_buffer + m_last_compression_point,
-            m_size - m_last_compression_point,
+            m_buffer,
+            m_size,
             &lz4Prefs
         );
 
         if (LZ4F_isError(compressedBytecount)) {
+            free(compressedBytes);
+
             throw std::runtime_error(
               std::string("Error compressing data using LZ4: ")
                 + LZ4F_getErrorName(compressedBytecount)
             );
         }
     }
 
-    m_size = m_last_compression_point;
+    m_size = 0;
 
     write<uint32_t>(compressedBytecount);
 
-    write_bytes((uint8_t*)compressedBytes, compressedBytecount, false);
+    write_bytes((uint8_t*)compressedBytes, compressedBytecount);
 
     free(compressedBytes);
 
-    m_last_compression_point = m_size;
+    m_compressed = true;
+}
+
+void SerializationBuffer::consolidate() {
+    if (m_wants_compress) {
+        for (auto blockPtr: m_blocks) {
+            blockPtr->compress();
+        }
+    }
+
+    if (m_blocks.size() == 1) {
+        return;
+    }
+
+    size_t totalSize = 0;
+
+    for (auto blockPtr: m_blocks) {
+        totalSize += blockPtr->size();
+    }
+
+    SerializationBufferBlock* block = new SerializationBufferBlock();
+    block->ensure(totalSize);
+
+    for (auto blockPtr: m_blocks) {
+        block->write_bytes(blockPtr->buffer(), blockPtr->size());
+    }
+
+    if (m_wants_compress) {
+        block->markCompressed();
+    }
+
+    m_blocks.clear();
+    m_blocks.push_back(
+        std::shared_ptr<SerializationBufferBlock>(
+            block
+        )
+    );
 }