Move event management code outside of the critical section

Author: flanglet

src/io/CompressedInputStream.cpp

@@ -727,7 +727,7 @@ void CompressedInputStream::close()
     // Force subsequent reads to trigger submitBlock immediately
     _bufferThreshold = 0;
 
-    // Buffer cleanup: force error on any subsequent write attempt
+    // Buffer cleanup: force error on any subsequent read attempt
     for (int i = 0; i < 2 * _jobs; i++) {
         if (_buffers[i]->_array != nullptr)
            delete[] _buffers[i]->_array;
@@ -773,6 +773,9 @@ template <class T>
 T DecodingTask<T>::run()
 {
     int blockId = _ctx.getInt("blockId");
+    bool streamPerTask = _ctx.getInt("tasks") > 1;
+    uint64 tType = _ctx.getLong("tType");
+    short eType = short(_ctx.getInt("eType"));
 
     // Lock free synchronization
     while (true) {
@@ -794,9 +797,6 @@ T DecodingTask<T>::run()
     EntropyDecoder* ed = nullptr;
     InputBitStream* ibs = nullptr;
     TransformSequence<byte>* transform = nullptr;
-    bool streamPerTask = _ctx.getInt("tasks") > 1;
-    uint64 tType = _ctx.getLong("tType");
-    short eType = short(_ctx.getInt("eType"));
 
     try {
         // Read shared bitstream sequentially (each task is gated by _processedBlockId)
@@ -816,11 +816,11 @@ T DecodingTask<T>::run()
             return T(*_data, blockId, 0, 0, Error::ERR_BLOCK_SIZE, "Invalid block size");
         }
 
-        const int r = int((read + 7) >> 3);
+        const uint r = uint((read + 7) >> 3);
 
         if (streamPerTask == true) {
-            if (_data->_length < max(_blockLength, r)) {
-                _data->_length = max(_blockLength, r);
+            if (_data->_length < int(max(_blockLength, r))) {
+                _data->_length = int(max(_blockLength, r));
                 delete[] _data->_array;
                 _data->_array = new byte[_data->_length];
             }
@@ -871,7 +871,8 @@ T DecodingTask<T>::run()
         const int length = dataSize << 3;
         const uint64 mask = (uint64(1) << length) - 1;
         const int preTransformLength = int(ibs->readBits(length) & mask);
-        const int maxTransformSize = min(max(_blockLength + _blockLength / 2, 2048), CompressedInputStream::MAX_BITSTREAM_BLOCK_SIZE);
+        const int maxTransformSize = int(min(max(_blockLength + _blockLength / 2, 2048u),
+                                         uint(CompressedInputStream::MAX_BITSTREAM_BLOCK_SIZE)));
 
         if ((preTransformLength <= 0) || (preTransformLength > maxTransformSize)) {
             // Error => cancel concurrent decoding tasks
@@ -911,7 +912,7 @@ T DecodingTask<T>::run()
             CompressedInputStream::notifyListeners(_listeners, evt2);
         }
 
-        const int bufferSize = max(_blockLength, preTransformLength + CompressedInputStream::EXTRA_BUFFER_SIZE);
+        const int bufferSize = max(int(_blockLength), preTransformLength + CompressedInputStream::EXTRA_BUFFER_SIZE);
 
         if (_buffer->_length < bufferSize) {
             _buffer->_length = bufferSize;

src/io/CompressedInputStream.hpp

@@ -110,7 +110,7 @@ namespace kanzi
    private:
        SliceArray<byte>* _data;
        SliceArray<byte>* _buffer;
-       int _blockLength;
+       uint _blockLength;
        DefaultInputBitStream* _ibs;
        XXHash32* _hasher32;
        XXHash64* _hasher64;

src/io/CompressedOutputStream.cpp

@@ -453,7 +453,9 @@ void CompressedOutputStream::processBuffer()
 #ifdef CONCURRENCY_ENABLED
     if (_futures[_bufferId].valid()) {
         EncodingTaskResult res = _futures[_bufferId].get();
-        if (res._error != 0) throw IOException(res._msg, res._error);
+
+        if (res._error != 0)
+            throw IOException(res._msg, res._error);
     }
 #endif
 
@@ -554,6 +556,7 @@ ostream& CompressedOutputStream::put(char c)
 #ifdef CONCURRENCY_ENABLED
             if (_futures[_bufferId].valid()) {
                 EncodingTaskResult res = _futures[_bufferId].get();
+
                 if (res._error != 0)
                     throw IOException(res._msg, res._error);
             }
@@ -792,6 +795,7 @@ T EncodingTask<T>::run()
         ee = nullptr;
         obs.close();
         uint64 written = obs.written();
+        const uint lw = (written < 8) ? 3 : uint(Global::log2(uint32(written >> 3)) + 4);
 
         // Lock free synchronization
         while (true) {
@@ -807,6 +811,22 @@ T EncodingTask<T>::run()
             CPU_PAUSE();
         }
 
+        // Emit block size in bits (max size pre-entropy is 1 GB = 1 << 30 bytes)
+        _obs->writeBits(lw - 3, 5); // write length-3 (5 bits max)
+        _obs->writeBits(written, lw);
+
+        // Emit data to shared bitstream
+        for (uint n = 0; written > 0; ) {
+            uint chkSize = uint(min(written, uint64(1) << 30));
+            _obs->writeBits(&_data->_array[n], chkSize);
+            n += ((chkSize + 7) >> 3);
+            written -= uint64(chkSize);
+        }
+
+        // After completion of the entropy coding, increment the block id.
+        // It unblocks the task processing the next block (if any).
+        _processedBlockId->store(blockId, memory_order_release);
+
         if (_listeners.size() > 0) {
             // Notify after entropy
             Event evt1(Event::AFTER_ENTROPY, blockId,
@@ -829,23 +849,6 @@ T EncodingTask<T>::run()
 #endif
         }
 
-        // Emit block size in bits (max size pre-entropy is 1 GB = 1 << 30 bytes)
-        const uint lw = (written < 8) ? 3 : uint(Global::log2(uint32(written >> 3)) + 4);
-        _obs->writeBits(lw - 3, 5); // write length-3 (5 bits max)
-        _obs->writeBits(written, lw);
-
-        // Emit data to shared bitstream
-        for (uint n = 0; written > 0; ) {
-            uint chkSize = uint(min(written, uint64(1) << 30));
-            _obs->writeBits(&_data->_array[n], chkSize);
-            n += ((chkSize + 7) >> 3);
-            written -= uint64(chkSize);
-        }
-
-        // After completion of the entropy coding, increment the block id.
-        // It unblocks the task processing the next block (if any).
-        _processedBlockId->store(blockId, memory_order_release);
-
         return T(blockId, 0, "Success");
     }
     catch (const exception& e) {