Skip to content

Commit 37ef082

Browse files
flangletflanglet
committed
Simplify code
1 parent 04ac904 commit 37ef082

2 files changed

Lines changed: 12 additions & 94 deletions

File tree

‎src/io/CompressedInputStream.cpp‎

Lines changed: 4 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -779,16 +779,18 @@ T DecodingTask<T>::run()
779779
bool streamPerTask = _ctx.getInt("tasks") > 1;
780780
uint64 tType = _ctx.getLong("tType");
781781
short eType = short(_ctx.getInt("eType"));
782-
#ifdef CONCURRENCY_ENABLED
783782
auto storeProcessedBlockId = [this](int value) {
783+
#ifdef CONCURRENCY_ENABLED
784784
{
785785
std::lock_guard<std::mutex> lock(*_blockMutex);
786786
STORE_ATOMIC(*_processedBlockId, value);
787787
}
788788

789789
_blockCondition->notify_all();
790-
};
790+
#else
791+
STORE_ATOMIC(*_processedBlockId, value);
791792
#endif
793+
};
792794

793795
#ifdef CONCURRENCY_ENABLED
794796
{
@@ -803,22 +805,6 @@ T DecodingTask<T>::run()
803805
// Skip, an error occurred
804806
return T(*_data, blockId, 0, 0, 0, "Canceled");
805807
}
806-
#else
807-
// Lock free synchronization
808-
while (true) {
809-
const int taskId = LOAD_ATOMIC(*_processedBlockId);
810-
811-
if (taskId == CompressedInputStream::CANCEL_TASKS_ID) {
812-
// Skip, an error occurred
813-
return T(*_data, blockId, 0, 0, 0, "Canceled");
814-
}
815-
816-
if (taskId == blockId - 1)
817-
break;
818-
819-
// Back-off improves performance
820-
CPU_PAUSE();
821-
}
822808
#endif
823809

824810
uint64 checksum1 = 0;
@@ -835,20 +821,12 @@ T DecodingTask<T>::run()
835821
uint64 read = _ibs->readBits(lr);
836822

837823
if (read == 0) {
838-
#ifdef CONCURRENCY_ENABLED
839824
storeProcessedBlockId(CompressedInputStream::CANCEL_TASKS_ID);
840-
#else
841-
STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
842-
#endif
843825
return T(*_data, blockId, 0, 0, 0, "Success");
844826
}
845827

846828
if (read > (uint64(1) << 34)) {
847-
#ifdef CONCURRENCY_ENABLED
848829
storeProcessedBlockId(CompressedInputStream::CANCEL_TASKS_ID);
849-
#else
850-
STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
851-
#endif
852830
return T(*_data, blockId, 0, 0, Error::ERR_BLOCK_SIZE, "Invalid block size");
853831
}
854832

@@ -876,11 +854,7 @@ T DecodingTask<T>::run()
876854

877855
// After completion of the bitstream reading, increment the block id.
878856
// It unblocks the task processing the next block (if any)
879-
#ifdef CONCURRENCY_ENABLED
880857
storeProcessedBlockId(blockId);
881-
#else
882-
STORE_ATOMIC(*_processedBlockId, blockId);
883-
#endif
884858

885859
// Check if the block must be skipped
886860
if (blockId < from) {
@@ -918,11 +892,7 @@ T DecodingTask<T>::run()
918892

919893
if ((preTransformLength <= 0) || (preTransformLength > maxTransformSize)) {
920894
// Error => cancel concurrent decoding tasks
921-
#ifdef CONCURRENCY_ENABLED
922895
storeProcessedBlockId(CompressedInputStream::CANCEL_TASKS_ID);
923-
#else
924-
STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
925-
#endif
926896
stringstream ss;
927897
ss << "Invalid compressed block length: " << preTransformLength;
928898

@@ -979,11 +949,7 @@ T DecodingTask<T>::run()
979949
// Block entropy decode
980950
if (ed->decode(_buffer->_array, 0, preTransformLength) != preTransformLength) {
981951
// Error => cancel concurrent decoding tasks
982-
#ifdef CONCURRENCY_ENABLED
983952
storeProcessedBlockId(CompressedInputStream::CANCEL_TASKS_ID);
984-
#else
985-
STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
986-
#endif
987953
delete ed;
988954

989955
if (streamPerTask == true)
@@ -1023,11 +989,7 @@ T DecodingTask<T>::run()
1023989
transform = nullptr;
1024990

1025991
if (res == false) {
1026-
#ifdef CONCURRENCY_ENABLED
1027992
storeProcessedBlockId(CompressedInputStream::CANCEL_TASKS_ID);
1028-
#else
1029-
STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
1030-
#endif
1031993
return T(*_data, blockId, 0, checksum1, Error::ERR_PROCESS_BLOCK,
1032994
"Transform inverse failed");
1033995
}
@@ -1039,11 +1001,7 @@ T DecodingTask<T>::run()
10391001
const uint32 checksum2 = _hasher32->hash(&_data->_array[savedIdx], decoded);
10401002

10411003
if (checksum2 != uint32(checksum1)) {
1042-
#ifdef CONCURRENCY_ENABLED
10431004
storeProcessedBlockId(CompressedInputStream::CANCEL_TASKS_ID);
1044-
#else
1045-
STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
1046-
#endif
10471005
stringstream ss;
10481006
ss << "Corrupted bitstream: expected checksum " << std::hex << checksum1 << ", found " << std::hex << checksum2;
10491007
return T(*_data, blockId, decoded, checksum1, Error::ERR_CRC_CHECK, ss.str());
@@ -1053,11 +1011,7 @@ T DecodingTask<T>::run()
10531011
const uint64 checksum2 = _hasher64->hash(&_data->_array[savedIdx], decoded);
10541012

10551013
if (checksum2 != checksum1) {
1056-
#ifdef CONCURRENCY_ENABLED
10571014
storeProcessedBlockId(CompressedInputStream::CANCEL_TASKS_ID);
1058-
#else
1059-
STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
1060-
#endif
10611015
stringstream ss;
10621016
ss << "Corrupted bitstream: expected checksum " << std::hex << checksum1 << ", found " << std::hex << checksum2;
10631017
return T(*_data, blockId, decoded, checksum1, Error::ERR_CRC_CHECK, ss.str());
@@ -1068,11 +1022,7 @@ T DecodingTask<T>::run()
10681022
}
10691023
catch (const exception& e) {
10701024
// Cancel any in-flight task waiting on this block.
1071-
#ifdef CONCURRENCY_ENABLED
10721025
storeProcessedBlockId(CompressedInputStream::CANCEL_TASKS_ID);
1073-
#else
1074-
STORE_ATOMIC(*_processedBlockId, CompressedInputStream::CANCEL_TASKS_ID);
1075-
#endif
10761026

10771027
if (transform != nullptr)
10781028
delete transform;

‎src/io/CompressedOutputStream.cpp‎

Lines changed: 8 additions & 40 deletions
Original file line numberDiff line numberDiff line change
@@ -626,34 +626,36 @@ T EncodingTask<T>::run()
626626
const int blockLength = _ctx.getInt("size");
627627
TransformSequence<kanzi::byte>* transform = nullptr;
628628
EntropyEncoder* ee = nullptr;
629-
#ifdef CONCURRENCY_ENABLED
630629
auto storeProcessedBlockId = [this](int value) {
630+
#ifdef CONCURRENCY_ENABLED
631631
{
632632
std::lock_guard<std::mutex> lock(*_blockMutex);
633633
STORE_ATOMIC(*_processedBlockId, value);
634634
}
635635

636636
_blockCondition->notify_all();
637+
#else
638+
STORE_ATOMIC(*_processedBlockId, value);
639+
#endif
637640
};
638641

639642
auto fetchAddProcessedBlockId = [this]() {
643+
#ifdef CONCURRENCY_ENABLED
640644
{
641645
std::lock_guard<std::mutex> lock(*_blockMutex);
642646
FETCH_ADD_ATOMIC(*_processedBlockId, 1);
643647
}
644648

645649
_blockCondition->notify_all();
646-
};
650+
#else
651+
FETCH_ADD_ATOMIC(*_processedBlockId, 1);
647652
#endif
653+
};
648654

649655
try {
650656
if (blockLength == 0) {
651657
// Last block (only block with 0 length)
652-
#ifdef CONCURRENCY_ENABLED
653658
fetchAddProcessedBlockId();
654-
#else
655-
FETCH_ADD_ATOMIC(*_processedBlockId, 1);
656-
#endif
657659
return T(blockId, 0, "Success");
658660
}
659661

@@ -743,23 +745,15 @@ T EncodingTask<T>::run()
743745
postTransformLength = _buffer->_index;
744746

745747
if (postTransformLength < 0) {
746-
#ifdef CONCURRENCY_ENABLED
747748
storeProcessedBlockId(CompressedOutputStream::CANCEL_TASKS_ID);
748-
#else
749-
STORE_ATOMIC(*_processedBlockId, CompressedOutputStream::CANCEL_TASKS_ID);
750-
#endif
751749
return T(blockId, Error::ERR_WRITE_FILE, "Invalid transform size");
752750
}
753751

754752
_ctx.putInt("size", postTransformLength);
755753
const int dataSize = (postTransformLength < 256) ? 1 : (Global::_log2(uint32(postTransformLength)) >> 3) + 1;
756754

757755
if (dataSize > 4) {
758-
#ifdef CONCURRENCY_ENABLED
759756
storeProcessedBlockId(CompressedOutputStream::CANCEL_TASKS_ID);
760-
#else
761-
STORE_ATOMIC(*_processedBlockId, CompressedOutputStream::CANCEL_TASKS_ID);
762-
#endif
763757
return T(blockId, Error::ERR_WRITE_FILE, "Invalid block data length");
764758
}
765759

@@ -822,11 +816,7 @@ T EncodingTask<T>::run()
822816
// Entropy encode block
823817
if (ee->encode(_buffer->_array, 0, postTransformLength) != postTransformLength) {
824818
delete ee;
825-
#ifdef CONCURRENCY_ENABLED
826819
storeProcessedBlockId(CompressedOutputStream::CANCEL_TASKS_ID);
827-
#else
828-
STORE_ATOMIC(*_processedBlockId, CompressedOutputStream::CANCEL_TASKS_ID);
829-
#endif
830820
return T(blockId, Error::ERR_PROCESS_BLOCK, "Entropy coding failed");
831821
}
832822

@@ -849,20 +839,6 @@ T EncodingTask<T>::run()
849839

850840
if (LOAD_ATOMIC(*_processedBlockId) == CompressedOutputStream::CANCEL_TASKS_ID)
851841
return T(blockId, 0, "Canceled");
852-
#else
853-
// Lock free synchronization
854-
while (true) {
855-
const int taskId = LOAD_ATOMIC(*_processedBlockId);
856-
857-
if (taskId == CompressedOutputStream::CANCEL_TASKS_ID)
858-
return T(blockId, 0, "Canceled");
859-
860-
if (taskId == blockId - 1)
861-
break;
862-
863-
// Back-off improves performance
864-
CPU_PAUSE();
865-
}
866842
#endif
867843

868844
// Emit block size in bits (max size pre-entropy is 1 GB = 1 << 30 bytes)
@@ -883,11 +859,7 @@ T EncodingTask<T>::run()
883859

884860
// After completion of the entropy coding, increment the block id.
885861
// It unblocks the task processing the next block (if any).
886-
#ifdef CONCURRENCY_ENABLED
887862
storeProcessedBlockId(blockId);
888-
#else
889-
STORE_ATOMIC(*_processedBlockId, blockId);
890-
#endif
891863

892864
if (_listeners.size() > 0) {
893865
// Notify after entropy
@@ -913,11 +885,7 @@ T EncodingTask<T>::run()
913885
}
914886
catch (const exception& e) {
915887
// Cancel any in-flight task waiting on this block.
916-
#ifdef CONCURRENCY_ENABLED
917888
storeProcessedBlockId(CompressedOutputStream::CANCEL_TASKS_ID);
918-
#else
919-
STORE_ATOMIC(*_processedBlockId, CompressedOutputStream::CANCEL_TASKS_ID);
920-
#endif
921889

922890
if (transform != nullptr)
923891
delete transform;

0 commit comments

Comments
 (0)