Index: /trunk/FACT++/src/fitsCompressor.cc =================================================================== --- /trunk/FACT++/src/fitsCompressor.cc (revision 16813) +++ /trunk/FACT++/src/fitsCompressor.cc (revision 16814) @@ -19,4 +19,29 @@ using namespace std; + +typedef struct TileHeader +{ + char id[4]; + uint32_t numRows; + uint64_t size; + TileHeader(uint32_t nRows=0, + uint64_t s=0) : id({'T', 'I', 'L', 'E'}), + numRows(nRows), + size(s) + { }; +} __attribute__((__packed__)) TileHeader; + +typedef struct BlockHeader +{ + uint64_t size; + char ordering; + unsigned char numProcs; + BlockHeader(uint64_t s=0, + char o=FACT_ROW_MAJOR, + unsigned char n=1) : size(s), + ordering(o), + numProcs(n) + {} +} __attribute__((__packed__)) BlockHeader; class CompressedFitsFile @@ -249,11 +274,13 @@ char t, int numOf, - uint32_t comp) : _name(n), - _num(numOf), - _typeSize(0), - _offset(0), - _type(t), - _description(""), - _compression(comp) + BlockHeader& head, + vector& seq) : _name(n), + _num(numOf), + _typeSize(0), + _offset(0), + _type(t), + _description(""), + _header(head), + _compSequence(seq) { switch (t) @@ -297,14 +324,22 @@ char type() const { return _type;}; string getDescription() const { return _description;} - uint32_t getCompression() const { return _compression;} + BlockHeader& getBlockHeader() { return _header;} + const vector& getCompressionSequence() const { return _compSequence;} + const char& getColumnOrdering() const { return _header.ordering;} + string getCompressionString() const { - switch (_compression) + return "FACT"; + /* + ostringstream str; + for (uint32_t i=0;i<_compSequence.size();i++) + switch (_compSequence[i]) { - case UNCOMPRESSED: return "RAW"; - case SMOOTHMAN: return "SMOO"; + case FACT_RAW: if (str.str().size() == 0) str << "RAW"; break; + case FACT_SMOOTHING: str << "SMOOTHING "; break; + case FACT_HUFFMAN16: str << "HUFFMAN16 "; break; }; - return "UNKNOWN"; + return str.str();*/ } @@ -317,5 +352,6 @@ char _type; ///< the type of the column, as specified by the fits documentation string _description; ///< a description for the column. It will be placed in the header - uint32_t _compression; ///< the compression of the column. Only looked for if used by the compressed writer + BlockHeader _header; + vector _compSequence; }; @@ -435,4 +471,5 @@ uint32_t compressHUFFMAN(char* dest, const char* src, uint32_t numRows, uint32_t sizeOfElems, uint32_t numRowElems); uint32_t compressSMOOTHMAN(char* dest, char* src, uint32_t numRows, uint32_t sizeOfElems, uint32_t numRowElems); + uint32_t applySMOOTHING(char* dest, char* src, uint32_t numRows, uint32_t sizeOfElems, uint32_t numRowElems); int32_t _checkOffset; ///< offset to the data pointer to calculate the checksum @@ -452,27 +489,4 @@ const uint32_t CompressedFitsFile::_THREAD_READ_ = 4; const uint32_t CompressedFitsFile::_THREAD_EXIT_ = 5; - -#define COMPRESSED_FLAG 0x1 -#define UNCOMPRESSED_FLAG 0x0 - -typedef struct TileHeader -{ - char id[4]; - uint32_t numRows; - uint64_t size; - TileHeader(uint32_t nRows=0, - uint64_t s=0) : id({'T', 'I', 'L', 'E'}), - numRows(nRows), - size(s) - { }; -} __attribute__((__packed__)) TileHeader; - -typedef struct BlockHeader -{ - uint64_t size; - char ordering; - unsigned char numProcs; - uint16_t procs[]; -} __attribute__((__packed__)) BlockHeader; template<> @@ -1140,8 +1154,7 @@ for (uint32_t i=0;i<_columns.size();i++) { - switch (_columns[i].getCompression()) - { - case UNCOMPRESSED: - case SMOOTHMAN: + switch (_columns[i].getColumnOrdering())//getCompression()) + { + case FACT_ROW_MAJOR: for (uint32_t k=0;k1;j--) + reinterpret_cast(src)[j] = reinterpret_cast(src)[j] - (reinterpret_cast(src)[j-1]+reinterpret_cast(src)[j-2])/2; + + return numRows*sizeOfElems*numRowElems; +} /**************************************************************** * COMPRESS BUFFER @@ -1262,36 +1286,62 @@ _catalog[currentCatalogRow][i].second = compressedOffset; - uint32_t compression = _columns[i].getCompression(); - if (_columns[i].numElems() == 0) continue; + + BlockHeader& head = _columns[i].getBlockHeader(); + const vector& sequence = _columns[i].getCompressionSequence(); //set the default byte telling if uncompressed the compressed Flag uint64_t previousOffset = compressedOffset; - _compressedBuffer[threadIndex][compressedOffset++] = COMPRESSED_FLAG; - switch (compression) - { - case UNCOMPRESSED: - compressedOffset += compressUNCOMPRESSED(&(_compressedBuffer[threadIndex][compressedOffset]), &(_transposedBuffer[threadIndex][offset]), thisRoundNumRows, _columns[i].sizeOfElems(), _columns[i].numElems()); - break; - case SMOOTHMAN: - compressedOffset += compressSMOOTHMAN(&(_compressedBuffer[threadIndex][compressedOffset]), &(_transposedBuffer[threadIndex][offset]), thisRoundNumRows, _columns[i].sizeOfElems(), _columns[i].numElems()); - break; - - default: - ; - }; + //skip header data + compressedOffset += sizeof(BlockHeader) + sizeof(uint16_t)*sequence.size(); + + for (uint32_t j=0;j _columns[i].sizeOfElems()*_columns[i].numElems()*thisRoundNumRows+1) + if (sequence[0] != FACT_RAW && + compressedOffset - previousOffset > _columns[i].sizeOfElems()*_columns[i].numElems()*thisRoundNumRows+sizeof(BlockHeader)+sizeof(uint16_t)*sequence.size()) {//if so set flag and redo it uncompressed - compressedOffset = previousOffset; - _compressedBuffer[threadIndex][compressedOffset++] = UNCOMPRESSED_FLAG; + cout << "REDOING UNCOMPRESSED" << endl; + compressedOffset = previousOffset + sizeof(BlockHeader) + 1; compressedOffset += compressUNCOMPRESSED(&(_compressedBuffer[threadIndex][compressedOffset]), &(_transposedBuffer[threadIndex][offset]), thisRoundNumRows, _columns[i].sizeOfElems(), _columns[i].numElems()); - } + BlockHeader he; + he.size = compressedOffset - previousOffset; + he.numProcs = 1; + he.ordering = FACT_ROW_MAJOR; + memcpy(&(_compressedBuffer[threadIndex][previousOffset]), (char*)(&he), sizeof(BlockHeader)); + _compressedBuffer[threadIndex][previousOffset+sizeof(BlockHeader)] = FACT_RAW; + offset += thisRoundNumRows*_columns[i].sizeOfElems()*_columns[i].numElems(); + _catalog[currentCatalogRow][i].first = compressedOffset - _catalog[currentCatalogRow][i].second; + continue; + } + head.size = compressedOffset - previousOffset; + memcpy(&(_compressedBuffer[threadIndex][previousOffset]), (char*)(&head), sizeof(BlockHeader)); + memcpy(&(_compressedBuffer[threadIndex][previousOffset+sizeof(BlockHeader)]), sequence.data(), sizeof(uint16_t)*sequence.size()); + offset += thisRoundNumRows*_columns[i].sizeOfElems()*_columns[i].numElems(); _catalog[currentCatalogRow][i].first = compressedOffset - _catalog[currentCatalogRow][i].second; } + TileHeader tHead(thisRoundNumRows, compressedOffset); -// tHead.numRows = thisRoundNumRows; -// tHead.size = compressedOffset; memcpy(_compressedBuffer[threadIndex], &tHead, sizeof(TileHeader)); return compressedOffset; @@ -1338,4 +1388,5 @@ pthread_mutex_unlock(&(myself->_mutex)); uint32_t threadToWaitForBeforeWriting = (myID == 0) ? myself->_numThreads-1 : myID-1; + while (myself->_threadStatus[myID] != _THREAD_EXIT_) { @@ -1603,4 +1654,14 @@ } + //get header structures + BlockHeader rawHeader; + BlockHeader smoothmanHeader(0, FACT_ROW_MAJOR, 3); + vector rawProcessings(1); + rawProcessings[0] = FACT_RAW; + vector smoothmanProcessings(3); + smoothmanProcessings[0] = FACT_SMOOTHING; + smoothmanProcessings[1] = FACT_HUFFMAN16; + smoothmanProcessings[2] = FACT_RAW; + //first lets see if we do have an explicit request bool explicitRequest = false; @@ -1612,7 +1673,7 @@ if (displayText) cout << compressions[j].second << endl; if (compressions[j].second == "UNCOMPRESSED") - outFile.addColumn(CompressedFitsFile::ColumnEntry(colName, sortedColumns[i].type, sortedColumns[i].num, CompressedFitsFile::UNCOMPRESSED)); + outFile.addColumn(CompressedFitsFile::ColumnEntry(colName, sortedColumns[i].type, sortedColumns[i].num, rawHeader, rawProcessings)); if (compressions[j].second == "SMOOTHMAN") - outFile.addColumn(CompressedFitsFile::ColumnEntry(colName, sortedColumns[i].type, sortedColumns[i].num, CompressedFitsFile::SMOOTHMAN)); + outFile.addColumn(CompressedFitsFile::ColumnEntry(colName, sortedColumns[i].type, sortedColumns[i].num, smoothmanHeader, smoothmanProcessings)); break; } @@ -1624,10 +1685,10 @@ { if (displayText) cout << "UNCOMPRESSED" << endl; - outFile.addColumn(CompressedFitsFile::ColumnEntry(colName, sortedColumns[i].type, sortedColumns[i].num, CompressedFitsFile::UNCOMPRESSED)); + outFile.addColumn(CompressedFitsFile::ColumnEntry(colName, sortedColumns[i].type, sortedColumns[i].num, rawHeader, rawProcessings)); } else { if (displayText) cout << "SMOOTHMAN" << endl; - outFile.addColumn(CompressedFitsFile::ColumnEntry(colName, sortedColumns[i].type, sortedColumns[i].num, CompressedFitsFile::SMOOTHMAN)); + outFile.addColumn(CompressedFitsFile::ColumnEntry(colName, sortedColumns[i].type, sortedColumns[i].num, smoothmanHeader, smoothmanProcessings)); } } Index: /trunk/Mars/mcore/zfits.h =================================================================== --- /trunk/Mars/mcore/zfits.h (revision 16813) +++ /trunk/Mars/mcore/zfits.h (revision 16814) @@ -13,4 +13,12 @@ #include "fits.h" #include "huffman.h" + + +#define FACT_RAW 0x0 +#define FACT_SMOOTHING 0x1 +#define FACT_HUFFMAN16 0x2 + +#define FACT_COL_MAJOR 'C' +#define FACT_ROW_MAJOR 'R' @@ -79,4 +87,6 @@ private: + + //Structure helper for reading tiles headers typedef struct TileHeader { @@ -84,15 +94,29 @@ uint32_t numRows; uint64_t size; - TileHeader(uint32_t nRows=0, - uint64_t s=0) : id({'T', 'I', 'L', 'E'}), + + TileHeader() {} + + TileHeader(uint32_t nRows, + uint64_t s) : id({'T', 'I', 'L', 'E'}), numRows(nRows), size(s) { }; - friend ostream& operator << (ostream& out, const TileHeader& h) - { - out << h.id[0] << h.id[1] << h.id[2] << h.id[3] << " num Rows: " << h.numRows << ", tile size: " << h.size; - return out; - } } __attribute__((__packed__)) TileHeader; + + //Structure helper for reading blocks headers and compresion schemes + typedef struct BlockHeader + { + uint64_t size; + char ordering; + unsigned char numProcs; + uint16_t processings[]; + + BlockHeader(uint64_t s=0, + char o=FACT_ROW_MAJOR, + unsigned char n=1) : size(s), + ordering(o), + numProcs(n) + {} + } __attribute__((__packed__)) BlockHeader; // Do what it takes to initialize the compressed structured @@ -103,4 +127,19 @@ return; + if (fTable.isCompressed) + for (auto it=fTable.sortedCols.begin(); it!= fTable.sortedCols.end(); it++) + if (it->comp != FACT) + { +#ifdef __EXCEPTIONS + throw runtime_error("ERROR: Only the FACT compression scheme is handled by this reader."); +#else + gLog << ___err___ << "ERROR: Only the FACT compression scheme is handled by this reader." << endl; + return; +#endif + } + + fColumnOrdering.resize(fTable.sortedCols.size()); + for (auto it=fColumnOrdering.begin(); it != fColumnOrdering.end(); it++) + (*it) = FACT_ROW_MAJOR; //Get compressed specific keywords fNumTiles = fTable.isCompressed ? GetInt("NAXIS2") : 0; @@ -129,4 +168,5 @@ vector fTransposedBuffer; /// fCompressedBuffer; /// fColumnOrdering; ///< ordering of the column's rows size_t fNumTiles; ///< Total number of tiles @@ -137,6 +177,6 @@ vector > > fCatalog;///< Catalog, i.e. the main table that points to the compressed data. - vector fTileSize; ///< size in bytes of each compressed tile - vector > fTileOffsets; ///< offset from start of tile of a given compressed column + vector fTileSize; ///< size in bytes of each compressed tile + vector > fTileOffsets; ///< offset from start of tile of a given compressed column void AllocateBuffers() @@ -148,5 +188,5 @@ fTransposedBuffer.resize(fTable.bytes_per_row*fNumRowsPerTile); - fCompressedBuffer.resize(fTable.bytes_per_row*fNumRowsPerTile + fTable.num_cols); //use a bit more memory for compression flags + fCompressedBuffer.resize(fTable.bytes_per_row*fNumRowsPerTile + fTable.num_cols*(sizeof(BlockHeader)+256)); //use a bit more memory for block headers } @@ -266,30 +306,38 @@ const char *src = fTransposedBuffer.data(); - for (auto it=fTable.sortedCols.begin(); it!=fTable.sortedCols.end(); it++) + uint32_t i=0; + for (auto it=fTable.sortedCols.begin(); it!=fTable.sortedCols.end(); it++, i++) { char *buffer = fBuffer.data() + it->offset; // pointer to column (destination buffer) - switch (it->comp) + switch (fColumnOrdering[i]) { - case UNCOMPRESSED: - case SMOOTHMAN: - // regular, "semi-transposed" copy - for (char *dest=buffer; destbytes); - src += it->bytes; // next column - } - break; - - default: - // transposed copy - for (char *elem=buffer; elembytes; elem+=it->size) // element-by-element (arrays) - { - for (char *dest=elem; destsize); - src += it->size; // next element + memcpy(dest, src, it->bytes); + src += it->bytes; // next column } - } + break; + + case FACT_COL_MAJOR: + // transposed copy + for (char *elem=buffer; elembytes; elem+=it->size) // element-by-element (arrays) + { + for (char *dest=elem; destsize); + src += it->size; // next element + } + } + break; + default: + #ifdef __EXCEPTIONS + throw runtime_error("Unkown column ordering scheme"); + #else + gLog << ___err___ << "ERROR - unkown column ordering scheme" << endl; + return; + #endif break; }; @@ -305,34 +353,25 @@ uint32_t UncompressUNCOMPRESSED(char* dest, const char* src, - uint32_t numRows, - uint32_t sizeOfElems, - uint32_t numRowElems) - { - memcpy(dest, src, numRows*sizeOfElems*numRowElems); - return numRows*sizeOfElems*numRowElems; + uint32_t numElems, + uint32_t sizeOfElems) + { + memcpy(dest, src, numElems*sizeOfElems); + return numElems*sizeOfElems; } // Read a bunch of data compressed with the Huffman algorithm - uint32_t UncompressHUFFMAN(char* dest, - const char* src, - uint32_t , - uint32_t sizeOfElems, - uint32_t numRowElems) - { - if (sizeOfElems < 2) - { - cout << "Error, Huffman only works on shorts or longer types. (here: " << sizeOfElems << "). Aborting." << endl; - return -1; - } - + uint32_t UncompressHUFFMAN16(char* dest, + const char* src, + uint32_t numChunks) + { vector uncompressed; //read compressed sizes (one per row) const uint32_t* compressedSizes = reinterpret_cast(src); - src += sizeof(uint32_t)*numRowElems; + src += sizeof(uint32_t)*numChunks; //uncompress the rows, one by one uint32_t sizeWritten = 0; - for (uint32_t j=0;j(src), compressedSizes[j], uncompressed); @@ -347,21 +386,13 @@ } - //Read a bunch of data compressed with the smoothman algorithm - uint32_t UncompressSMOOTHMAN(int16_t* dest, - const char* src, - uint32_t numRows, - uint32_t sizeOfElems, - uint32_t numRowElems) - { - //call huffman transposed - const uint32_t sizeWritten = UncompressHUFFMAN(reinterpret_cast(dest), src, numRowElems, sizeOfElems, numRows); - + uint32_t UnApplySMOOTHING(int16_t* data, + uint32_t numElems) + { //un-do the integer smoothing - for (uint32_t j=2;j(dest), src, thisRoundNumRows, col.size, col.num); - break; - - default: - ; + const int64_t compressedOffset = fTileOffsets[catalogCurrentRow][i]; + + BlockHeader* head = reinterpret_cast(&fCompressedBuffer[compressedOffset]); + + fColumnOrdering[i] = head->ordering; + + uint32_t numRows = (head->ordering==FACT_ROW_MAJOR) ? thisRoundNumRows : col.num; + uint32_t numCols = (head->ordering==FACT_COL_MAJOR) ? thisRoundNumRows : col.num; + + const char *src = fCompressedBuffer.data()+compressedOffset+sizeof(BlockHeader)+sizeof(uint16_t)*head->numProcs; + + for (uint32_t j=head->numProcs;j != 0; j--) + { + uint32_t sizeWritten=0; + + switch (head->processings[j-1]) + { + case FACT_RAW: + if (head->numProcs == 1) + sizeWritten = UncompressUNCOMPRESSED(dest, src, numRows*numCols, col.size); + break; + case FACT_SMOOTHING: + sizeWritten = UnApplySMOOTHING(reinterpret_cast(dest), numRows*numCols); + break; + case FACT_HUFFMAN16: + sizeWritten = UncompressHUFFMAN16(dest, src, numRows); + break; + default: +#ifdef __EXCEPTIONS + throw runtime_error("Unknown processing applied to data. Aborting"); +#else + gLog << ___err___ << "ERROR - Unknown processing applied to data. Aborting" << endl; + return; +#endif + break; + } + //increment destination counter only when processing done. + if (j==1) dest+= sizeWritten; } }