Changeset 16814 for trunk/FACT++/src

Timestamp:

06/12/13 15:56:12 (11 years ago)

Author:

lyard

Message:

added block headers and changed compression mechanism

File:

• trunk/FACT++/src/fitsCompressor.cc (modified) (15 diffs)

trunk/FACT++/src/fitsCompressor.cc

@@ -19 +19 @@
 
 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 +274 @@
                         char          t,
                         int           numOf,
-                        uint32_t      comp) : _name(n),
-                                              _num(numOf),
-                                              _typeSize(0),
-                                              _offset(0),
-                                              _type(t),
-                                              _description(""),
-                                              _compression(comp)
+                        BlockHeader&   head,
+                        vector<uint16_t>& seq) : _name(n),
+                                                 _num(numOf),
+                                                 _typeSize(0),
+                                                 _offset(0),
+                                                 _type(t),
+                                                 _description(""),
+                                                 _header(head),
+                                                 _compSequence(seq)
             {
                 switch (t)
@@ -297 +324 @@
             char          type()                 const { return _type;};
             string        getDescription()       const { return _description;}
-            uint32_t      getCompression()       const { return _compression;}
+            BlockHeader& getBlockHeader()  { return _header;}
+            const vector<uint16_t>& 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 +352 @@
             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<uint16_t> _compSequence;
     };
 
@@ -435 +471 @@
         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 +489 @@
 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 +1154 @@
     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;k<thisRoundNumRows;k++)
                 {//regular, "semi-transposed" copy
@@ -1150 +1163 @@
                 }
             break;
-            default :
+
+            case FACT_COL_MAJOR :
                 for (int j=0;j<_columns[i].numElems();j++)
                     for (uint32_t k=0;k<thisRoundNumRows;k++)
@@ -1158 +1172 @@
                     }
             break;
+            default:
+                    cout << "Error: unknown column ordering: " << _columns[i].getColumnOrdering() << endl;
 
         };
@@ -1247 +1263 @@
 }
 
+uint32_t CompressedFitsWriter::applySMOOTHING(char* , char* src, uint32_t numRows, uint32_t sizeOfElems, uint32_t numRowElems)
+{
+    uint32_t colWidth = numRowElems;
+    for (int j=colWidth*numRows-1;j>1;j--)
+        reinterpret_cast<int16_t*>(src)[j] = reinterpret_cast<int16_t*>(src)[j] - (reinterpret_cast<int16_t*>(src)[j-1]+reinterpret_cast<int16_t*>(src)[j-2])/2;
+
+    return numRows*sizeOfElems*numRowElems;
+}
 /****************************************************************
  *                  COMPRESS BUFFER
@@ -1262 +1286 @@
         _catalog[currentCatalogRow][i].second = compressedOffset;
 
-        uint32_t compression = _columns[i].getCompression();
-
         if (_columns[i].numElems() == 0) continue;
+
+        BlockHeader& head = _columns[i].getBlockHeader();
+        const vector<uint16_t>& 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<sequence.size(); j++)
+        {
+            switch (sequence[j])
+            {
+                case FACT_RAW:
+                    if (head.numProcs == 1)
+                        compressedOffset += compressUNCOMPRESSED(&(_compressedBuffer[threadIndex][compressedOffset]), &(_transposedBuffer[threadIndex][offset]), thisRoundNumRows, _columns[i].sizeOfElems(), _columns[i].numElems());
+                break;
+                case FACT_SMOOTHING:
+                        applySMOOTHING(&(_compressedBuffer[threadIndex][compressedOffset]), &(_transposedBuffer[threadIndex][offset]), thisRoundNumRows, _columns[i].sizeOfElems(), _columns[i].numElems());
+                break;
+                case FACT_HUFFMAN16:
+                    if (head.ordering == FACT_COL_MAJOR)
+                        compressedOffset += compressHUFFMAN(&(_compressedBuffer[threadIndex][compressedOffset]), &(_transposedBuffer[threadIndex][offset]), thisRoundNumRows, _columns[i].sizeOfElems(), _columns[i].numElems());
+                    else
+                        compressedOffset += compressHUFFMAN(&(_compressedBuffer[threadIndex][compressedOffset]), &(_transposedBuffer[threadIndex][offset]), _columns[i].numElems(), _columns[i].sizeOfElems(),  thisRoundNumRows);
+                break;
+                default:
+                    cout << "ERROR: Unkown compression sequence entry: " << sequence[i] << endl;
+                break;
+            }
+        }
+
         //check if compressed size is larger than uncompressed
-        if (compression != UNCOMPRESSED &&
-            compressedOffset - previousOffset > _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 +1388 @@
     pthread_mutex_unlock(&(myself->_mutex));
     uint32_t threadToWaitForBeforeWriting = (myID == 0) ? myself->_numThreads-1 : myID-1;
+
     while (myself->_threadStatus[myID] != _THREAD_EXIT_)
     {
@@ -1603 +1654 @@
         }
 
+        //get header structures
+        BlockHeader rawHeader;
+        BlockHeader smoothmanHeader(0, FACT_ROW_MAJOR, 3);
+        vector<uint16_t> rawProcessings(1);
+        rawProcessings[0] = FACT_RAW;
+        vector<uint16_t> 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 +1673 @@
                 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 +1685 @@
         {
             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));
         }
     }