Index: trunk/Mars/mcore/zofits.h =================================================================== --- trunk/Mars/mcore/zofits.h (revision 17257) +++ trunk/Mars/mcore/zofits.h (revision 17258) @@ -58,7 +58,10 @@ } + WriteTarget() { } + WriteTarget(const WriteTarget &t, uint32_t sz) : tile_num(t.tile_num), size(sz), data(t.data) { } + uint32_t tile_num; ///< Tile index of the data (to make sure that they are written in the correct order) uint32_t size; ///< Size to write - shared_ptr data; ///< Memory block to write + shared_ptr data; ///< Memory block to write }; @@ -119,11 +122,8 @@ void InitMemberVariables(const uint32_t nt=0, const uint32_t rpt=0, const uint64_t maxUsableMem=0) { - if (nt == 0) - throw runtime_error("There must be at least 1 tile of data (0 specified). This is required by the FITS standard. Please try again with num_tile >= 1."); - fCheckOffset = 0; fNumQueues = 0; - fNumTiles = nt; + fNumTiles = nt==0 ? 1 : nt; fNumRowsPerTile = rpt; @@ -136,4 +136,5 @@ fThreadsException = exception_ptr(); #endif + fErrno = 0; } @@ -206,10 +207,11 @@ { const uint32_t one_catalog_row_size = fTable.num_cols*2*sizeof(uint64_t); - const uint32_t total_catalog_size = fNumTiles*one_catalog_row_size;//fCatalog.size()*one_catalog_row_size; + const uint32_t total_catalog_size = fNumTiles*one_catalog_row_size; // swap the catalog bytes before writing vector swapped_catalog(total_catalog_size); + uint32_t shift = 0; - for (auto it=fCatalog.begin(); it!=fCatalog.end(); it++) + for (auto it=fCatalog.cbegin(); it!=fCatalog.cend(); it++) { revcpy(swapped_catalog.data() + shift, (char*)(it->data()), fTable.num_cols*2); @@ -259,5 +261,9 @@ return fCatalog.back(); } + /// write one row of data + /// Note, in a multi-threaded environment (NumThreads>0), the return code should be checked rather + /// than the badbit() of the stream (it might have been set by a thread before the errno has been set) + /// errno will then contain the correct error number of the last error which happened during writing. /// @param ptr the source buffer /// @param the number of bytes to write @@ -275,4 +281,10 @@ } +#ifdef __EXCEPTIONS + //check if something hapenned while the compression threads were working + //if so, re-throw the exception that was generated + if (fThreadsException != exception_ptr()) + rethrow_exception(fThreadsException); +#endif //copy current row to pool or rows waiting for compression char* target_location = fSmartBuffer.get() + fRealRowWidth*(fTable.num_rows%fNumRowsPerTile); @@ -305,43 +317,46 @@ DrsOffsetCalibrate(target_location); - if (fTable.num_rows % fNumRowsPerTile == 0) - { - CompressionTarget compress_target(AddOneCatalogRow()); - SetNextCompression(compress_target); - - if (fNumQueues == 0) - { //no worker threads. do everything in-line - const uint64_t size_to_write = CompressBuffer(compress_target); - - WriteTarget write_target; - write_target.size = size_to_write; - write_target.data = compress_target.target.data; - write_target.tile_num = compress_target.target.tile_num; - - return WriteBufferToDisk(write_target); + if (fTable.num_rows % fNumRowsPerTile != 0) + { + errno = fErrno; + return errno==0; + } + + CompressionTarget compress_target(AddOneCatalogRow()); + SetNextCompression(compress_target); + + //no worker threads. do everything in-line + if (fNumQueues == 0) + { + const uint64_t size_to_write = CompressBuffer(compress_target); + + const WriteTarget write_target(compress_target.target, size_to_write); + if (!WriteBufferToDisk(write_target)) + throw runtime_error("Unexpected tile number mismatch in WriteBufferToDisk in the main thread."); + + // The correct 'errno' is set, because it is the main thread. + return good(); + } + + //if all queues are empty, use queue 0 + uint32_t min_index = 0; + uint32_t min_size = numeric_limits::max(); + uint32_t current_index = 0; + + for (auto it=fCompressionQueues.cbegin(); it!=fCompressionQueues.cend(); it++) + { + if (it->size() < min_size) + { + min_index = current_index; + min_size = it->size(); } - else - { - //if all queues are empty, use queue 0 - uint32_t min_index = 0; - uint32_t min_size = numeric_limits::max(); - uint32_t current_index = 0; - - for (auto it=fCompressionQueues.cbegin(); it!=fCompressionQueues.cend(); it++) - { - if (it->size() < min_size) - { - min_index = current_index; - min_size = it->size(); - } - current_index++; - } - - if (!fCompressionQueues[min_index].post(compress_target)) - throw runtime_error("The compression queues are not started. Did you close the file before writing this row ?"); - } - } - - return good(); + current_index++; + } + + if (!fCompressionQueues[min_index].emplace(compress_target)) + throw runtime_error("The compression queues are not started. Did you close the file before writing this row?"); + + errno = fErrno; + return errno==0; } @@ -358,18 +373,14 @@ void SetNextCompression(CompressionTarget& target) { - //get space for transposed data - shared_ptr transposed_data = fMemPool.malloc(); + //fill up compression target + target.src = fSmartBuffer; + target.transposed_src = fMemPool.malloc(); + target.num_rows = fTable.num_rows; //fill up write to disk target - WriteTarget write_target; + WriteTarget &write_target = target.target; write_target.tile_num = (fTable.num_rows-1)/fNumRowsPerTile; write_target.size = 0; write_target.data = fMemPool.malloc(); - - //fill up compression target - target.src = fSmartBuffer; - target.transposed_src = transposed_data; - target.target = write_target; - target.num_rows = fTable.num_rows; //get a new buffer to host the incoming data @@ -434,19 +445,11 @@ if (tellp() < 0) - { -#ifdef __EXCEPTIONS - throw runtime_error("Looks like the file has been closed already"); -#else return false; -#endif - } #ifdef __EXCEPTIONS //check if something hapenned while the compression threads were working + //if so, re-throw the exception that was generated if (fThreadsException != exception_ptr()) - { - //if so, re-throw the exception that was generated rethrow_exception(fThreadsException); - } #endif @@ -458,16 +461,13 @@ //set number of threads to zero before calling compressBuffer - int32_t backup_num_queues = fNumQueues; + const int32_t backup_num_queues = fNumQueues; fNumQueues = 0; - uint64_t size_to_write = CompressBuffer(compress_target); + + const uint64_t size_to_write = CompressBuffer(compress_target); fNumQueues = backup_num_queues; - WriteTarget write_target; - write_target.size = size_to_write; - write_target.data = compress_target.target.data; - write_target.tile_num = compress_target.target.tile_num; - + const WriteTarget write_target(compress_target.target, size_to_write); if (!WriteBufferToDisk(write_target)) - throw runtime_error("Something went wrong while writing the last tile..."); + throw runtime_error("Tile number mismatch in WriteBufferToDisk writing the last tile."); } @@ -504,8 +504,5 @@ SetInt("NAXIS1", total_catalog_width); SetInt("NAXIS2", total_num_tiles_written); - - ostringstream str; - str << fRawSum.val(); - SetStr("RAWSUM", str.str()); + SetStr("RAWSUM", to_string(fRawSum.val())); const float compression_ratio = (float)(fRealRowWidth*fTable.num_rows)/(float)heap_size; @@ -551,7 +548,7 @@ return false; + const size_t size = SizeFromType(typechar); + Table::Column col; - size_t size = SizeFromType(typechar); - col.name = name; col.type = typechar; @@ -562,19 +559,11 @@ fRealRowWidth += size*cnt; - fRealColumns.emplace_back(CompressedColumn(col, comp)); - - ostringstream strKey, strVal, strCom; - strKey << "ZFORM" << fRealColumns.size(); - strVal << cnt << typechar; - strCom << "format of " << name << " [" << CommentFromType(typechar); - SetStr(strKey.str(), strVal.str(), strCom.str()); - - strKey.str(""); - strVal.str(""); - strCom.str(""); - strKey << "ZCTYP" << fRealColumns.size(); - strVal << "FACT"; - strCom << "Compression type FACT"; - SetStr(strKey.str(), strVal.str(), strCom.str()); + fRealColumns.emplace_back(col, comp); + + ostringstream str; + str << "format of " << name << " [" << CommentFromType(typechar); + + SetStr("ZFORM"+to_string(fRealColumns.size()), to_string(cnt)+typechar, str.str()); + SetStr("ZCTYP"+to_string(fRealColumns.size()), "FACT", "Compression type: FACT"); return true; @@ -639,14 +628,18 @@ int32_t extraBytes = 0; uint32_t sizeToChecksum = sizeToWrite; + + //should we extend the array to the left ? if (fCheckOffset != 0) - {//should we extend the array to the left ? - sizeToChecksum += fCheckOffset; + { + sizeToChecksum += fCheckOffset; checkSumPointer -= fCheckOffset; memset(checkSumPointer, 0, fCheckOffset); } + + //should we extend the array to the right ? if (sizeToChecksum%4 != 0) - {//should we extend the array to the right ? + { extraBytes = 4 - (sizeToChecksum%4); - memset(checkSumPointer+sizeToChecksum, 0,extraBytes); + memset(checkSumPointer+sizeToChecksum, 0, extraBytes); sizeToChecksum += extraBytes; } @@ -705,10 +698,5 @@ //post the result to the writing queue //get a copy so that it becomes non-const - WriteTarget wt; - wt.tile_num = target.target.tile_num; - wt.size = compressed_size; - wt.data = target.target.data; - - fWriteToDiskQueue.post(wt); + fWriteToDiskQueue.emplace(target.target, compressed_size); // if used by the queue, always return true as the elements are not ordered @@ -730,23 +718,10 @@ { #endif + //could not write the data to disk if (!writeCompressedDataToDisk(target.data.get(), target.size)) - {//could not write the data to disk - ostringstream str; - str << "An error occured while writing to disk: "; - if (eof()) - str << "End-Of-File"; - if (failbit) - str << "Logical error on i/o operation"; - if (badbit) - str << "Writing error on i/o operation"; -#ifdef __EXCEPTIONS - throw runtime_error(str.str()); -#else - gLog << ___err___ << "ERROR - " << str.str() << endl; -#endif - } -#ifdef __EXCEPTIONS - } - catch(...) + fErrno = errno; +#ifdef __EXCEPTIONS + } + catch (...) { fThreadsException = current_exception(); @@ -780,5 +755,6 @@ catalog_row[i].second = compressedOffset; - if (fRealColumns[i].col.num == 0) continue; + if (fRealColumns[i].col.num == 0) + continue; Compression& head = fRealColumns[i].block_head; @@ -794,15 +770,17 @@ switch (head.getProc(j)) { - case kFactRaw: - compressedOffset += compressUNCOMPRESSED(dest + compressedOffset, src + offset, thisRoundNumRows*fRealColumns[i].col.size*fRealColumns[i].col.num); + case kFactRaw: + compressedOffset += compressUNCOMPRESSED(dest + compressedOffset, src + offset, thisRoundNumRows*fRealColumns[i].col.size*fRealColumns[i].col.num); break; - case kFactSmoothing: - applySMOOTHING(src + offset, thisRoundNumRows*fRealColumns[i].col.num); + + case kFactSmoothing: + applySMOOTHING(src + offset, thisRoundNumRows*fRealColumns[i].col.num); break; - case kFactHuffman16: - if (head.getOrdering() == kOrderByCol) - compressedOffset += compressHUFFMAN16(dest + compressedOffset, src + offset, thisRoundNumRows, fRealColumns[i].col.size, fRealColumns[i].col.num); - else - compressedOffset += compressHUFFMAN16(dest + compressedOffset, src + offset, fRealColumns[i].col.num, fRealColumns[i].col.size, thisRoundNumRows); + + case kFactHuffman16: + if (head.getOrdering() == kOrderByCol) + compressedOffset += compressHUFFMAN16(dest + compressedOffset, src + offset, thisRoundNumRows, fRealColumns[i].col.size, fRealColumns[i].col.num); + else + compressedOffset += compressHUFFMAN16(dest + compressedOffset, src + offset, fRealColumns[i].col.num, fRealColumns[i].col.size, thisRoundNumRows); break; } @@ -837,5 +815,5 @@ } - TileHeader tile_head(thisRoundNumRows, compressedOffset); + const TileHeader tile_head(thisRoundNumRows, compressedOffset); memcpy(dest, &tile_head, sizeof(TileHeader)); @@ -855,19 +833,21 @@ switch (fRealColumns[i].block_head.getOrdering()) { - case kOrderByRow: + case kOrderByRow: + //regular, "semi-transposed" copy + for (uint32_t k=0;k(data); - //un-do the integer smoothing for (uint32_t j=2;j> fCompressionQueues; ///< Processing queues (=threads) - Queue> fWriteToDiskQueue; ///< Writing queue (=thread) + Queue> fWriteToDiskQueue; ///< Writing queue (=thread) // catalog related stuff CatalogType fCatalog; ///< Catalog for this file - uint32_t fCatalogSize; ///< Actual catalog size (.size() is slow on large lists) - uint32_t fNumTiles; ///< Number of pre-reserved tiles - uint32_t fNumRowsPerTile; ///< Number of rows per tile - off_t fCatalogOffset; ///< Offset of the catalog from the beginning of the file + uint32_t fCatalogSize; ///< Actual catalog size (.size() is slow on large lists) + uint32_t fNumTiles; ///< Number of pre-reserved tiles + uint32_t fNumRowsPerTile; ///< Number of rows per tile + off_t fCatalogOffset; ///< Offset of the catalog from the beginning of the file // checksum related stuff Checksum fCatalogSum; ///< Checksum of the catalog - Checksum fRawSum; ///< Raw sum (specific to FACT) - int32_t fCheckOffset; ///< offset to the data pointer to calculate the checksum + Checksum fRawSum; ///< Raw sum (specific to FACT) + int32_t fCheckOffset; ///< offset to the data pointer to calculate the checksum // data layout related stuff @@ -986,6 +968,6 @@ block_head(h) {} - Table::Column col; ///< the regular column entry - Compression block_head; ///< the compression data associated with that column + Table::Column col; ///< the regular column entry + Compression block_head; ///< the compression data associated with that column }; vector fRealColumns; ///< Vector hosting the columns of the file @@ -997,4 +979,6 @@ exception_ptr fThreadsException; ///< exception pointer to store exceptions coming from the threads #endif + int fErrno; ///< propagate errno to main thread + };