Index: /trunk/Mars/mcore/factofits.h =================================================================== --- /trunk/Mars/mcore/factofits.h (revision 17236) +++ /trunk/Mars/mcore/factofits.h (revision 17237) @@ -11,4 +11,8 @@ #include "zofits.h" +#include "../src/DataWriteFits2.h" + +#include "DrsCalib.h" + #ifndef __MARS__ namespace std @@ -23,4 +27,5 @@ public: + /// constructors factofits(uint32_t numTiles=1000, uint32_t rowPerTile=100, @@ -30,4 +35,5 @@ fDataOffset = -1; } + factofits(const char* fname, uint32_t numTiles=1000, @@ -38,9 +44,10 @@ fDataOffset = -1; } + virtual ~factofits() { } - //whether or not a calibration was given to the file writer + /// whether or not a calibration was given to the file writer virtual bool IsOffsetCalibration() { @@ -48,7 +55,9 @@ } - //assign a given drs offset calibration + ///assign a given drs offset calibration void SetDrsCalibration(const vector &calib) { + VerifyCalibrationSize(calib.size()); + if (!IsOffsetCalibration()) fOffsetCalibration.resize(1440*1024); @@ -58,17 +67,9 @@ } - void SetDrsCalibration(const vector& calib) - { - if (calib.size() != 1440*1024) - #ifdef __EXCEPTIONS - throw runtime_error("Cannot load calibration with anything else than 1024 samples per pixel"); - #else - gLog << ___err___ << "ERROR - Cannot load calibration with anything else than 1024 samples per pixel"); - #endif - SetDrsCalibration(calib.data()); - } - + ///assign a given drs offset calibration void SetDrsCalibration(const vector& vec) { + VerifyCalibrationSize(vec.size()); + if (!IsOffsetCalibration()) fOffsetCalibration.resize(1440*1024); @@ -78,4 +79,5 @@ } + ///assign a given drs offset calibration void SetDrsCalibration(const DrsCalibration& drs) { @@ -83,4 +85,6 @@ return; + VerifyCalibrationSize(drs.fOffset.size()); + if (!IsOffsetCalibration()) fOffsetCalibration.resize(1440*1024); @@ -90,4 +94,5 @@ } + ///Overload of the super function bool WriteTableHeader(const char* name="DATA") { @@ -102,7 +107,8 @@ if (it->col.name == "StartCellData") fStartCellsOffset = it->col.offset; + if (it->col.name == "Data") { - fNumSlices = it->col.num; + fNumSlices = it->col.num; fDataOffset = it->col.offset; if (fNumSlices % 1440 != 0) @@ -118,4 +124,5 @@ } } + if (fStartCellsOffset < 0) { @@ -128,4 +135,5 @@ fOffsetCalibration.resize(0); } + if (fDataOffset < 0) { @@ -143,4 +151,5 @@ } + ///Actually write the drs calibration table virtual bool WriteDrsOffsetsTable() { @@ -191,10 +200,11 @@ } + ///Apply the drs offset calibration (overload of super-method) virtual void DrsOffsetCalibrate(char* target_location) { if (IsOffsetCalibration()) { - int16_t* startCell = reinterpret_cast(target_location + fStartCellsOffset); - int16_t* data = reinterpret_cast(target_location + fDataOffset); + const int16_t* startCell = reinterpret_cast(target_location + fStartCellsOffset); + int16_t* data = reinterpret_cast(target_location + fDataOffset); for (uint32_t ch=0; ch<1440; ch++) @@ -226,9 +236,24 @@ private: + /// Checks if the size of the input calibration is ok + void VerifyCalibrationSize(uint32_t size) + { + if (size != 1440*1024) + { + ostringstream str; + str << "Cannot load calibration with anything else than 1440 pixels and 1024 samples per pixel. Got a total size of " << size; +#ifdef __EXCEPTIONS + throw runtime_error(str.str()); +#else + gLog << ___err___ << "ERROR - " << str.str() << endl; +#endif + } + } + //Offsets calibration stuff. - vector fOffsetCalibration; ///< The calibration itself - int32_t fStartCellsOffset; ///< Offset in bytes for the startcell data - int32_t fDataOffset; ///< Offset in bytes for the data - int32_t fNumSlices; ///< Number of samples per pixel per event + vector fOffsetCalibration; ///< The calibration itself + int32_t fStartCellsOffset; ///< Offset in bytes for the startcell data + int32_t fDataOffset; ///< Offset in bytes for the data + int32_t fNumSlices; ///< Number of samples per pixel per event }; //class factofits Index: /trunk/Mars/mcore/zofits.h =================================================================== --- /trunk/Mars/mcore/zofits.h (revision 17236) +++ /trunk/Mars/mcore/zofits.h (revision 17237) @@ -22,7 +22,7 @@ #endif - class zofits : public ofits { + /// Overriding of the begin() operator to get the smallest item in the list instead of the true begin template struct QueueMin : std::list @@ -34,4 +34,5 @@ }; + /// Parameters required to write a tile to disk struct WriteTarget { @@ -41,25 +42,23 @@ } - uint32_t tile_num; - uint32_t size; - shared_ptr target; + 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 }; + /// Parameters required to compress a tile of data struct CompressionTarget { - /* - bool operator < (const CompressionTarget& other) - { - return target < other.target; - }*/ - - shared_ptr src; - shared_ptr transposed_src; - WriteTarget target; - uint32_t num_rows; + shared_ptr src; ///< Original data + shared_ptr transposed_src; ///< Transposed data + WriteTarget target; ///< Compressed data + uint32_t num_rows; ///< Number of rows to compress }; public: - //constructors + /// constructors + /// @param numTiles how many data groups should be pre-reserved ? + /// @param rowPerTile how many rows will be grouped together in a single tile + /// @param maxUsableMem how many bytes of memory can be used by the compression buffers zofits(uint32_t numTiles=1000, uint32_t rowPerTile=100, @@ -72,4 +71,8 @@ } + /// @param fname the target filename + /// @param numTiles how many data groups should be pre-reserved ? + /// @param rowPerTile how many rows will be grouped together in a single tile + /// @param maxUsableMem how many bytes of memory can be used by the compression buffers zofits(const char* fname, uint32_t numTiles=1000, @@ -83,4 +86,5 @@ } + /// destructors virtual ~zofits() { @@ -88,19 +92,22 @@ //initialization of member variables - void InitMemberVariables(uint32_t nt=0, uint32_t rpt=0, uint64_t maxUsableMem=0) + /// @param nt number of tiles + /// @param rpt number of rows per tile + /// @param maxUsableMem max amount of RAM to be used by the compression buffers + void InitMemberVariables(const uint32_t nt=0, const uint32_t rpt=0, const uint64_t maxUsableMem=0) { if (nt == 0) - throw runtime_error("Cannot work with a catalog of size 0. sorry."); - - fCheckOffset = 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; fNumRowsPerTile = rpt; - fBuffer = NULL; - fRealRowWidth = 0; + fBuffer = NULL; + fRealRowWidth = 0; fCatalogExtraRows = 0; - - fCatalogOffset = 0; + fCatalogOffset = 0; fMaxUsableMem = maxUsableMem; @@ -110,10 +117,10 @@ } - - //write the header of the binary table + /// write the header of the binary table + /// @param name the name of the table to be created + /// @return the state of the file virtual bool WriteTableHeader(const char* name="DATA") { - if (!reallocateBuffers()) - throw ("While allocating memory: apparently there not as much free memory as advertized..."); + reallocateBuffers(); ofits::WriteTableHeader(name); @@ -125,4 +132,5 @@ it->start(); + //start the disk writer fWriteToDiskQueue.start(); } @@ -134,4 +142,7 @@ } + /// open a new file. + /// @param filename the name of the file + /// @param Whether or not the name of the extension should be added or not void open(const char* filename, bool addEXTNAMEKey=true) { @@ -139,13 +150,13 @@ //add compression-related header entries - SetBool("ZTABLE", true, "Table is compressed"); - SetInt("ZNAXIS1", 0, "Width of uncompressed rows"); - SetInt("ZNAXIS2", 0, "Number of uncompressed rows"); - SetInt("ZPCOUNT", 0, ""); - SetInt("ZHEAPPTR", 0, ""); - SetInt("ZTILELEN", fNumRowsPerTile, "Number of rows per tile"); - SetInt("THEAP", 0, ""); - SetStr("RAWSUM", " 0", "Checksum of raw little endian data"); - SetFloat("ZRATIO", 0, "Compression ratio"); + SetBool( "ZTABLE", true, "Table is compressed"); + SetInt( "ZNAXIS1", 0, "Width of uncompressed rows"); + SetInt( "ZNAXIS2", 0, "Number of uncompressed rows"); + SetInt( "ZPCOUNT", 0, ""); + SetInt( "ZHEAPPTR", 0, ""); + SetInt( "ZTILELEN", fNumRowsPerTile, "Number of rows per tile"); + SetInt( "THEAP", 0, ""); + SetStr( "RAWSUM", " 0", "Checksum of raw little endian data"); + SetFloat("ZRATIO", 0, "Compression ratio"); fCatalogExtraRows = 0; @@ -153,4 +164,6 @@ } + /// Super method. does nothing as zofits does not know about DrsOffsets + /// @return the state of the file virtual bool WriteDrsOffsetsTable() { @@ -158,4 +171,6 @@ } + /// Returns the number of bytes per uncompressed row + /// @return number of bytes per uncompressed row uint32_t GetBytesPerRow() const { @@ -163,9 +178,12 @@ } + /// Write the data catalog + /// @return the state of the file bool WriteCatalog() { const uint32_t one_catalog_row_size = fTable.num_cols*2*sizeof(uint64_t); - const uint32_t total_catalog_size = fCatalog.size()*one_catalog_row_size; - + const uint32_t total_catalog_size = fCatalog.size()*one_catalog_row_size; + + // swap the catalog bytes before writing vector swapped_catalog(total_catalog_size); uint32_t shift = 0; @@ -176,11 +194,12 @@ } + // first time writing ? remember where we are if (fCatalogOffset == 0) - { fCatalogOffset = tellp(); - } - + + // remember where we came from const off_t where_are_we = tellp(); + // write to disk seekp(fCatalogOffset); write(swapped_catalog.data(), total_catalog_size); @@ -188,4 +207,5 @@ seekp(where_are_we); + // udpate checksum fCatalogSum.reset(); fCatalogSum.add(swapped_catalog.data(), total_catalog_size); @@ -193,4 +213,6 @@ return good(); } + + /// Applies the DrsOffsets calibration to the data. Does nothing as zofits knows nothing about drsoffsets. virtual void DrsOffsetCalibrate(char* ) { @@ -198,4 +220,5 @@ } + /// Grows the catalog in case not enough rows were allocated void GrowCatalog() { @@ -214,4 +237,8 @@ } + /// write one row of data + /// @param ptr the source buffer + /// @param the number of bytes to write + /// @return the state of the file. WARNING: with multithreading, this will most likely be the state of the file before the data is actually written bool WriteRow(const void* ptr, size_t cnt, bool = true) { @@ -242,5 +269,5 @@ //for now, make an extra copy of the data, for RAWSUM checksuming. - //Ideally this should be moved to the threads, along with the drs-offset-calibration + //Ideally this should be moved to the threads //However, because the RAWSUM must be calculated before the tile is transposed, I am not sure whether //one extra memcpy per row written is worse than 100 rows checksumed when the tile is full.... @@ -278,9 +305,8 @@ WriteTarget write_target; write_target.size = size_to_write; - write_target.target = compress_target.target.target; + write_target.data = compress_target.target.data; write_target.tile_num = compress_target.target.tile_num; - if (!WriteBufferToDisk(write_target)) - throw runtime_error("Something went wrong while writing to disk"); + return WriteBufferToDisk(write_target); } else @@ -302,5 +328,5 @@ if (!fCompressionQueues[min_index].post(compress_target)) - throw runtime_error("I could not post this buffer. This does not make sense..."); + throw runtime_error("The compression queues are not started. Did you close the file before writing this row ?"); } } @@ -309,11 +335,14 @@ } + /// update the real number of rows void FlushNumRows() { - SetInt("NAXIS2", fTable.num_rows/fNumRowsPerTile); + SetInt("NAXIS2", (fTable.num_rows + fNumRowsPerTile-1)/fNumRowsPerTile); SetInt("ZNAXIS2", fTable.num_rows); FlushHeader(); } + /// Setup the environment to compress yet another tile of data + /// @param target the struct where to host the produced parameters void SetNextCompression(CompressionTarget& target) { @@ -325,11 +354,11 @@ write_target.tile_num = (fTable.num_rows-1)/fNumRowsPerTile; write_target.size = 0; - write_target.target = fMemPool.malloc(); + 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; + target.transposed_src = transposed_data; + target.target = write_target; + target.num_rows = fTable.num_rows; //get a new buffer to host the incoming data @@ -338,4 +367,5 @@ } + /// Shrinks a catalog that is too long to fit into the reserved space at the beginning of the file. void ShrinkCatalog() { @@ -390,6 +420,9 @@ } + /// close an open file. + /// @return the state of the file bool close() { + // stop compression and write threads for (auto it=fCompressionQueues.begin(); it != fCompressionQueues.end(); it++) it->wait(); @@ -400,5 +433,5 @@ { #ifdef __EXCEPTIONS - throw runtime_error("Something went wrong while writing to disk..."); + throw runtime_error("Looks like the file has been closed already"); #else return false; @@ -407,11 +440,13 @@ #ifdef __EXCEPTIONS - //check if something hapenned to the compression threads + //check if something hapenned while the compression threads were working if (fThreadsException != exception_ptr()) { + //if so, re-throw the exception that was generated rethrow_exception(fThreadsException); } #endif + //write the last tile of data (if any if (fTable.num_rows%fNumRowsPerTile != 0) { @@ -427,5 +462,5 @@ WriteTarget write_target; write_target.size = size_to_write; - write_target.target = compress_target.target.target; + write_target.data = compress_target.target.data; write_target.tile_num = compress_target.target.tile_num; @@ -440,12 +475,11 @@ SetInt("ZNAXIS2", fTable.num_rows); - uint64_t heap_offset = fCatalog.size()*fTable.num_cols*sizeof(uint64_t)*2; - SetInt("ZHEAPPTR", heap_offset); + SetInt("ZHEAPPTR", fCatalog.size()*fTable.num_cols*sizeof(uint64_t)*2); const uint32_t total_num_tiles_written = (fTable.num_rows + fNumRowsPerTile-1)/fNumRowsPerTile; - - SetInt("THEAP", total_num_tiles_written*2*sizeof(int64_t)*fTable.num_cols); - - SetInt("NAXIS1", 2*sizeof(int64_t)*fTable.num_cols); + const uint32_t total_catalog_width = 2*sizeof(int64_t)*fTable.num_cols; + + SetInt("THEAP", total_num_tiles_written*total_catalog_width); + SetInt("NAXIS1", total_catalog_width); SetInt("NAXIS2", total_num_tiles_written); @@ -471,5 +505,5 @@ } - float compression_ratio = (float)(fRealRowWidth*fTable.num_rows)/(float)heap_size; + const float compression_ratio = (float)(fRealRowWidth*fTable.num_rows)/(float)heap_size; SetFloat("ZRATIO", compression_ratio); @@ -478,5 +512,4 @@ SetInt("PCOUNT", heap_size, "size of special data area"); - //Just for updating the fCatalogSum value @@ -502,5 +535,5 @@ } - //Overload of the ofits method. Just calls the zofits specific one with default, uncompressed options for this column + /// Overload of the ofits method. Just calls the zofits specific one with default, uncompressed options for this column bool AddColumn(uint32_t cnt, char typechar, const string& name, const string& unit, const string& comment="", bool addHeaderKeys=true) { @@ -508,4 +541,5 @@ } + /// Overload of the simplified compressed version bool AddColumn(const FITS::Compression &comp, uint32_t cnt, char typechar, const string& name, const string& unit, const string& comment="", bool addHeaderKeys=true) { @@ -543,8 +577,10 @@ } + /// static setter for the default number of threads to use. -1 means all available physical cores static void SetDefaultNumThreads(int32_t num) { fgNumQueues = num;} static int32_t GetDefaultNumThreads() { return fgNumQueues;} - int32_t GetNumThreads() { return fNumQueues;} + /// Get and set the actual number of threads for this object + int32_t GetNumThreads() const { return fNumQueues;} bool SetNumThreads(int32_t num) { @@ -554,8 +590,10 @@ throw runtime_error("File must be closed before changing the number of compression threads"); #else - gLog << ___err___ << "ERROR - File must be closed before changing the number of compression threads"; + gLog << ___err___ << "ERROR - File must be closed before changing the number of compression threads" << endl; #endif return false; } + + //get number of physically available threads #ifdef USE_BOOST_THREADS int32_t num_available_cores = boost::thread::hardware_concurrency(); @@ -564,23 +602,23 @@ #endif + // could not detect number of available cores from system properties... + // assume that 5 cores are available (4 compression, 1 write) if (num_available_cores == 0) - {//could not detect number of available cores from system properties... - //Assuming that 5 cores are availables (4 compression, 1 write) num_available_cores = 5; - } + + // Throw an exception if too many cores are requested if (num > num_available_cores) { ostringstream str; - str << "Number of threads cannot be greater than physically available (" << num_available_cores << ")"; + str << "You will be using more threads(" << num << ") than available cores(" << num_available_cores << "). Expect sub-optimal performances"; #ifdef __EXCEPTIONS throw runtime_error(str.str()); #else - gLog << ___err___ << "ERROR - " << str.str(); -#endif - return false; + gLog << ___err___ << "WARNING - " << str.str() << endl; +#endif } if (num == -1) - num = num_available_cores-2; // 1 for writing, one for the main thread + num = num_available_cores-2; // 1 for writing, 1 for the main thread if (fCompressionQueues.size() == (uint32_t)num) @@ -590,5 +628,5 @@ Queue queue(bind(&zofits::CompressBuffer, this, placeholders::_1), false); - //shrink + //shrink if required if ((uint32_t)num < fCompressionQueues.size()) { @@ -597,5 +635,5 @@ } - //grow + //grow if required fCompressionQueues.resize(num, queue); @@ -607,7 +645,8 @@ protected: - bool reallocateBuffers() - { - size_t chunk_size = fRealRowWidth*fNumRowsPerTile + fRealColumns.size()*sizeof(BlockHeader) + sizeof(TileHeader) + 8; //+8 for checksuming; + /// Allocates the required objects. + void reallocateBuffers() + { + const size_t chunk_size = fRealRowWidth*fNumRowsPerTile + fRealColumns.size()*sizeof(BlockHeader) + sizeof(TileHeader) + 8; //+8 for checksuming; fMemPool.setChunkSize(chunk_size); @@ -625,8 +664,11 @@ *it = CatalogEntry(0,0); } - return true; - } - - bool writeCompressedDataToDisk(char* src, uint32_t sizeToWrite) + } + + /// Actually does the writing to disk (and checksuming) + /// @param src the buffer to write + /// @param sizeToWrite how many bytes should be written + /// @return the state of the file + bool writeCompressedDataToDisk(char* src, const uint32_t sizeToWrite) { char* checkSumPointer = src+4; @@ -656,4 +698,7 @@ } + /// Compress a given buffer based on the target. This is the method executed by the threads + /// @param target the struct hosting the parameters of the compression + /// @return number of bytes of the compressed data, or always 1 when used by the Queues uint32_t CompressBuffer(const CompressionTarget& target) { @@ -667,5 +712,5 @@ //compress the buffer - compressed_size = compressBuffer(target.target.target.get()->get(), target.transposed_src.get()->get(), target.num_rows); + compressed_size = compressBuffer(target.target.data.get()->get(), target.transposed_src.get()->get(), target.num_rows); #ifdef __EXCEPTIONS } @@ -686,11 +731,14 @@ wt.tile_num = target.target.tile_num; wt.size = compressed_size; - wt.target = target.target.target; + wt.data = target.target.data; fWriteToDiskQueue.post(wt); - return compressed_size; - } - + // if used by the queue, always return true as the elements are not ordered + return 1; + } + + /// Write one compressed tile to disk. This is the method executed by the writing thread + /// @param target the struct hosting the write parameters bool WriteBufferToDisk(const WriteTarget& target) { @@ -701,14 +749,47 @@ fLatestWrittenTile++; - //write the buffer to disk. - return writeCompressedDataToDisk(target.target.get()->get(), target.size); - } - +#ifdef __EXCEPTIONS + try + { +#endif + if (!writeCompressedDataToDisk(target.data.get()->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(...) + { + fThreadsException = current_exception(); + if (fNumQueues == 0) + rethrow_exception(fThreadsException); + } +#endif + return true; + } + + /// Compress a given buffer based on its source and destination //src cannot be const, as applySMOOTHING is done in place + /// @param dest the buffer hosting the compressed data + /// @param src the buffer hosting the transposed data + /// @param num_rows the number of uncompressed rows in the transposed buffer + /// @param the number of bytes of the compressed data uint64_t compressBuffer(char* dest, char* src, uint32_t num_rows) { - uint32_t thisRoundNumRows = (num_rows%fNumRowsPerTile) ? num_rows%fNumRowsPerTile : fNumRowsPerTile; - uint32_t offset=0; - uint32_t currentCatalogRow = (num_rows-1)/fNumRowsPerTile; + const uint32_t thisRoundNumRows = (num_rows%fNumRowsPerTile) ? num_rows%fNumRowsPerTile : fNumRowsPerTile; + const uint32_t currentCatalogRow = (num_rows-1)/fNumRowsPerTile; + uint32_t offset = 0; //skip the checksum reserved area @@ -728,5 +809,5 @@ //set the default byte telling if uncompressed the compressed Flag - uint64_t previousOffset = compressedOffset; + const uint64_t previousOffset = compressedOffset; //skip header data @@ -755,5 +836,5 @@ if ((head.getProc(0) != kFactRaw) && (compressedOffset - previousOffset > fRealColumns[i].col.size*fRealColumns[i].col.num*thisRoundNumRows+head.getSizeOnDisk()))// && two) {//if so set flag and redo it uncompressed - cout << "Redoing uncompressed ! " << endl; + // cout << "Redoing uncompressed ! " << endl; //de-smooth ! if (head.getProc(0) == kFactSmoothing) @@ -787,7 +868,10 @@ } + /// Transpose a tile to a new buffer + /// @param src buffer hosting the regular, row-ordered data + /// @param dest the target buffer that will receive the transposed data void copyTransposeTile(const char* src, char* dest) { - uint32_t thisRoundNumRows = (fTable.num_rows%fNumRowsPerTile) ? fTable.num_rows%fNumRowsPerTile : fNumRowsPerTile; + const uint32_t thisRoundNumRows = (fTable.num_rows%fNumRowsPerTile) ? fTable.num_rows%fNumRowsPerTile : fNumRowsPerTile; //copy the tile and transpose it @@ -817,4 +901,8 @@ /// Specific compression functions + /// @param dest the target buffer + /// @param src the source buffer + /// @param size number of bytes to copy + /// @return number of bytes written uint32_t compressUNCOMPRESSED(char* dest, const char* src, uint32_t size) { @@ -823,4 +911,11 @@ } + /// Do huffman encoding + /// @param dest the buffer that will receive the compressed data + /// @param src the buffer hosting the transposed data + /// @param numRows number of rows of data in the transposed buffer + /// @param sizeOfElems size in bytes of one data elements + /// @param numRowElems number of elements on each row + /// @return number of bytes written uint32_t compressHUFFMAN16(char* dest, const char* src, uint32_t numRows, uint32_t sizeOfElems, uint32_t numRowElems) { @@ -836,5 +931,5 @@ throw runtime_error("HUFMANN16 can only encode columns with 16-bit or longer types"); #else - gLog << ___err___ << "ERROR - HUFMANN16 can only encode columns with 16-bit or longer types"; + gLog << ___err___ << "ERROR - HUFMANN16 can only encode columns with 16-bit or longer types" << endl; return 0; #endif @@ -859,5 +954,9 @@ } - uint32_t applySMOOTHING(char* data, uint32_t numElems)//uint32_t numRows, uint32_t sizeOfElems, uint32_t numRowElems) + /// Applies Thomas' DRS4 smoothing + /// @param data where to apply it + /// @param numElems how many elements of type int16_t are stored in the buffer + /// @return number of bytes modified + uint32_t applySMOOTHING(char* data, uint32_t numElems) { int16_t* short_data = reinterpret_cast(data); @@ -867,6 +966,10 @@ return numElems*sizeof(int16_t); } - // Apply the inverse transform of the integer smoothing - uint32_t UnApplySMOOTHING(char* data, uint32_t numElems) + + /// Apply the inverse transform of the integer smoothing + /// @param data where to apply it + /// @param numElems how many elements of type int16_t are stored in the buffer + /// @return number of bytes modified + uint32_t UnApplySMOOTHING(char* data, uint32_t numElems) { int16_t* short_data = reinterpret_cast(data); @@ -877,44 +980,40 @@ return numElems*sizeof(uint16_t); } - //Compressed data stuff - int32_t fCheckOffset; ///< offset to the data pointer to calculate the checksum - uint32_t fNumTiles; - uint32_t fNumRowsPerTile; - - MemoryManager fMemPool; + + //thread related stuff - vector> fCompressionQueues; - Queue> fWriteToDiskQueue; - - //thread related stuff - static int32_t fgNumQueues; ///< The number of threads that will be used to compress - int32_t fNumQueues; ///< The number of threads that will be used to compress - - int32_t fLatestWrittenTile; -#ifdef __EXCEPTIONS - exception_ptr fThreadsException; -#endif + MemoryManager fMemPool; ///< Actual memory manager, providing memory for the compression buffers + static int32_t fgNumQueues; ///< Default number of threads to be used by the objects + int32_t fNumQueues; ///< Current number of threads that will be used by this object + uint64_t fMaxUsableMem; ///< Maximum number of bytes that can be allocated by the memory manager + int32_t fLatestWrittenTile; ///< Index of the last tile written to disk (for correct ordering while using several threads) + + vector> fCompressionQueues; ///< Processing queues (=threads) + Queue> fWriteToDiskQueue; ///< Writing queue (=thread) + + // catalog related stuff struct CatalogEntry { CatalogEntry(int64_t f=0, int64_t s=0) : first(f), second(s) {}; - int64_t first; - int64_t second; + int64_t first; ///< Size of this column in the tile + int64_t second; ///< offset of this column in the tile, from the start of the heap area } __attribute__((__packed__)); - typedef vector CatalogRow; - typedef vector CatalogType; - CatalogType fCatalog; - Checksum fCatalogSum; - Checksum fRawSum; - off_t fCatalogOffset; - uint32_t fRealRowWidth; - uint32_t fCatalogExtraRows; - vector fRawSumBuffer; - uint64_t fMaxUsableMem; - - shared_ptr fSmartBuffer; - char* fBuffer; - + typedef vector CatalogRow; + typedef vector CatalogType; + CatalogType fCatalog; ///< Catalog for this file + 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 fCatalogExtraRows; ///< Number of extra rows written on top of the initial capacity 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 + + // data layout related stuff + /// Regular columns augmented with compression informations struct CompressedColumn { @@ -922,8 +1021,16 @@ block_head(h) {} - Table::Column col; - Compression block_head; + Table::Column col; ///< the regular column entry + Compression block_head; ///< the compression data associated with that column }; - vector fRealColumns; + vector fRealColumns; ///< Vector hosting the columns of the file + uint32_t fRealRowWidth; ///< Width in bytes of one uncompressed row + shared_ptr fSmartBuffer; ///< Smart pointer to the buffer where the incoming rows are written + char* fBuffer; ///< regular version of fSmartBuffer + vector fRawSumBuffer;///< buffer used for checksuming the incoming data, before compression + +#ifdef __EXCEPTIONS + exception_ptr fThreadsException; ///< exception pointer to store exceptions coming from the threads +#endif };