Changeset 17253 for trunk/Mars/mcore/zofits.h

Timestamp:

10/18/13 13:49:17 (11 years ago)

Author:

lyard

Message:

zofits with catalog shrinking and compressed calibration table

File:

• trunk/Mars/mcore/zofits.h (modified) (24 diffs)

trunk/Mars/mcore/zofits.h

@@ -37 +37 @@
         };
 
+        //catalog types
+        struct CatalogEntry
+        {
+            CatalogEntry(int64_t f=0, int64_t s=0) : first(f), second(s) {};
+            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<CatalogEntry> CatalogRow;
+        typedef list<CatalogRow>     CatalogType;
+
         /// Parameters required to write a tile to disk
         struct WriteTarget
@@ -53 +63 @@
         struct CompressionTarget
         {
-            shared_ptr<MemoryChunk> src;             ///< Original data
-            shared_ptr<MemoryChunk> transposed_src; ///<  Transposed data
-            WriteTarget             target;        ///<   Compressed data
-            uint32_t                num_rows;     ///<    Number of rows to compress
-        };
+            CompressionTarget(CatalogRow& r) : catalog_entry(r)
+            {}
+
+            CatalogRow&             catalog_entry;   ///< Reference to the catalog entry to deal with
+            shared_ptr<MemoryChunk> src;            ///<  Original data
+            shared_ptr<MemoryChunk> transposed_src;///<   Transposed data
+            WriteTarget             target;       ///<    Compressed data
+            uint32_t                num_rows;    ///<     Number of rows to compress
+         };
 
 public:
@@ -112 +126 @@
             fBuffer           = NULL;
             fRealRowWidth     = 0;
-            fCatalogExtraRows = 0;
             fCatalogOffset    = 0;
+            fCatalogSize      = 0;
 
             fMaxUsableMem = maxUsableMem;
@@ -163 +177 @@
             SetStr(  "RAWSUM",   "         0",    "Checksum of raw little endian data");
             SetFloat("ZRATIO",   0,               "Compression ratio");
-
-            fCatalogExtraRows = 0;
+            SetInt(  "ZSHRINK",  1,               "Catalog shrink factor");
+
+            fCatalogSize      = 0;
+            fCatalog.clear();
             fRawSum.reset();
         }
@@ -187 +203 @@
         {
             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   = fNumTiles*one_catalog_row_size;//fCatalog.size()*one_catalog_row_size;
 
             // swap the catalog bytes before writing
@@ -198 +214 @@
             }
 
+            if (fCatalogSize < fNumTiles)
+                memset(swapped_catalog.data()+shift, 0, total_catalog_size-shift);
+
             // first time writing ? remember where we are
             if (fCatalogOffset == 0)
@@ -208 +227 @@
             seekp(fCatalogOffset);
             write(swapped_catalog.data(), total_catalog_size);
+
             if (where_are_we != fCatalogOffset)
                 seekp(where_are_we);
@@ -224 +244 @@
         }
 
-        /// Grows the catalog in case not enough rows were allocated
-        void GrowCatalog()
-        {
-            uint32_t orig_catalog_size = fCatalog.size();
-
-            fCatalog.resize(fCatalog.size()*2);
-            for (uint32_t i=orig_catalog_size;i<fCatalog.size(); i++)
-            {
-                fCatalog[i].resize(fTable.num_cols);
-                for (auto it=(fCatalog[i].begin()); it!=fCatalog[i].end(); it++)
-                    *it = CatalogEntry(0,0);
-            }
-
-            fCatalogExtraRows += orig_catalog_size;
-            fNumTiles         += orig_catalog_size;
-        }
-
+        CatalogRow& AddOneCatalogRow()
+        {
+            // add one row to the catalog
+            fCatalog.emplace_back(CatalogRow());
+            fCatalog.back().resize(fTable.num_cols);
+            for (auto it=fCatalog.back().begin(); it != fCatalog.back().end(); it++)
+                *it = CatalogEntry(0,0);
+
+            fCatalogSize++;
+
+            return fCatalog.back();
+        }
         /// write one row of data
         /// @param ptr the source buffer
@@ -253 +268 @@
 #else
                 gLog << ___err___ << "ERROR - Wrong size of row given to WriteRow" << endl;
-                return false;
-#endif
-            }
-
-            if (fTable.num_rows >= fNumRowsPerTile*fNumTiles)
-            {
-//                GrowCatalog();
-#ifdef __EXCEPTIONS
-                throw runtime_error("Maximum number of rows exceeded for this file");
-#else
-                gLog << ___err___ << "ERROR - Maximum number of rows exceeded for this file" << endl;
                 return false;
 #endif
@@ -294 +298 @@
             fRawSum.add(fRawSumBuffer, false);
 
+            fTable.num_rows++;
+
             DrsOffsetCalibrate(target_location);
 
-            fTable.num_rows++;
-
             if (fTable.num_rows % fNumRowsPerTile == 0)
             {
-                CompressionTarget compress_target;
+                CompressionTarget compress_target(AddOneCatalogRow());
                 SetNextCompression(compress_target);
 
@@ -374 +378 @@
         void ShrinkCatalog()
         {
+            //add empty row to get either the target number of rows, or a multiple of the allowed size
+            for (uint32_t i=0;i<fCatalogSize%fNumTiles;i++)
+                AddOneCatalogRow();
+
             //did we write more rows than what the catalog could host ?
-            if (fCatalogExtraRows != 0)
-            {
-                //how many rows can the regular catalog host ?
-                const uint32_t max_regular_rows = (fCatalog.size() - fCatalogExtraRows)*fNumRowsPerTile;
-                //what's the shrink factor to be applied ?
-                const uint32_t shrink_factor = fTable.num_rows/max_regular_rows + ((fTable.num_rows%max_regular_rows) ? 1 : 0);
+            if (fCatalogSize > fNumTiles)
+            {
+                const uint32_t shrink_factor = fCatalogSize / fNumTiles; //always exact as extra rows were added just above
 
                 //shrink the catalog !
-                for (uint32_t i=0; i<fTable.num_rows/fNumRowsPerTile; i+= shrink_factor)
-                {//add the elements one by one, so that the empty ones at the end (i.e. fTable.num_rows%shrink_factor) do not create havok
-                    const uint32_t target_catalog_row = i/shrink_factor;
-                    //move data from current row (i) to target row
-                    for (uint32_t j=0; j<fTable.num_cols; j++)
-                    {
-                        fCatalog[target_catalog_row][j].second = fCatalog[i][j].second;
-                        fCatalog[target_catalog_row][j].first  = 0;
-                        uint64_t last_size   = fCatalog[i][j].first;
-                        uint64_t last_offset = fCatalog[i][j].second;
-
-                        for (uint32_t k=1; k<shrink_factor; k++)
-                        {
-                           if (fCatalog[i+k][j].second != 0)
-                           {
-                               fCatalog[target_catalog_row][j].first +=  fCatalog[i+k][j].second - last_offset;
-                           }
-                           else
-                           {
-                               fCatalog[target_catalog_row][j].first += last_size;
-                               break;
-                           }
-                           last_size   = fCatalog[i+k][j].first;
-                           last_offset = fCatalog[i+k][j].second;
-                        }
-                    }
+                uint32_t entry_id = 1;
+                auto it = fCatalog.begin();
+                it++;
+                for (; it != fCatalog.end(); it++)
+                {
+                    if (entry_id >= fNumTiles) break;
+
+                    uint32_t target_id = entry_id*shrink_factor;
+
+                    auto jt = it;
+                    for (uint32_t i=0;i<target_id-entry_id;i++)
+                        jt++;
+
+                    *it = *jt;
+
+                    entry_id++;
                 }
 
-                fCatalog.resize(fCatalog.size() - fCatalogExtraRows);
-
+                const uint32_t num_tiles_to_remove = fCatalogSize-fNumTiles;
+                //remove the too many entries
+                for (uint32_t i=0;i<num_tiles_to_remove;i++)
+                {
+                    fCatalog.pop_back();
+                    fCatalogSize--;
+                }
                 //update header keywords
-                const uint32_t new_num_rows_per_tiles = fNumRowsPerTile*shrink_factor;
-                const uint32_t new_num_tiles_written = (fTable.num_rows + new_num_rows_per_tiles-1)/new_num_rows_per_tiles;
-                SetInt("THEAP", new_num_tiles_written*2*sizeof(int64_t)*fTable.num_cols);
-                SetInt("NAXIS2", new_num_tiles_written);
-                SetInt("ZTILELEN", new_num_rows_per_tiles);
-                cout << "New num rows per tiles: " << new_num_rows_per_tiles << " shrink factor: " << shrink_factor << endl;
-                cout << "Num tiles written: " << new_num_tiles_written << endl;
+                fNumRowsPerTile *= shrink_factor;
+
+                SetInt("ZTILELEN", fNumRowsPerTile);
+                SetInt("ZSHRINK",  shrink_factor);
             }
         }
@@ -455 +452 @@
             if (fTable.num_rows%fNumRowsPerTile != 0)
             {
-                CompressionTarget compress_target;
+                CompressionTarget compress_target(AddOneCatalogRow());
                 SetNextCompression(compress_target);
 
@@ -475 +472 @@
             AlignTo2880Bytes();
 
+            int64_t heap_size = 0;
+            int64_t compressed_offset = 0;
+            for (auto it=fCatalog.begin(); it!= fCatalog.end(); it++)
+            {
+                compressed_offset += sizeof(TileHeader);
+                heap_size         += sizeof(TileHeader);
+                for (uint32_t j=0; j<it->size(); j++)
+                {
+                    heap_size += (*it)[j].first;
+                    (*it)[j].second = compressed_offset;
+                    compressed_offset += (*it)[j].first;
+                    if ((*it)[j].first == 0)
+                        (*it)[j].second = 0;
+                }
+            }
+
+            ShrinkCatalog();
+
             //update header keywords
             SetInt("ZNAXIS1", fRealRowWidth);
             SetInt("ZNAXIS2", fTable.num_rows);
 
-            SetInt("ZHEAPPTR", fCatalog.size()*fTable.num_cols*sizeof(uint64_t)*2);
+            SetInt("ZHEAPPTR", fCatalogSize*fTable.num_cols*sizeof(uint64_t)*2);
 
             const uint32_t total_num_tiles_written = (fTable.num_rows + fNumRowsPerTile-1)/fNumRowsPerTile;
@@ -492 +507 @@
             SetStr("RAWSUM", str.str());
 
-            int64_t heap_size = 0;
-            int64_t compressed_offset = 0;
-
-            for (uint32_t i=0; i<total_num_tiles_written; i++)
-            {
-                compressed_offset += sizeof(TileHeader);
-                heap_size         += sizeof(TileHeader);
-                for (uint32_t j=0; j<fCatalog[i].size(); j++)
-                {
-                    heap_size += fCatalog[i][j].first;
-                    fCatalog[i][j].second = compressed_offset;
-                    compressed_offset += fCatalog[i][j].first;
-                    if (fCatalog[i][j].first == 0)
-                        fCatalog[i][j].second = 0;
-                }
-            }
-
             const float compression_ratio = (float)(fRealRowWidth*fTable.num_rows)/(float)heap_size;
             SetFloat("ZRATIO", compression_ratio);
 
             //add to the heap size the size of the gap between the catalog and the actual heap
-            heap_size += (fCatalog.size() - total_num_tiles_written)*fTable.num_cols*sizeof(uint64_t)*2;
+            heap_size += (fCatalogSize - total_num_tiles_written)*fTable.num_cols*sizeof(uint64_t)*2;
 
             SetInt("PCOUNT", heap_size, "size of special data area");
@@ -630 +628 @@
 
             fRawSumBuffer.resize(fRealRowWidth + 4-fRealRowWidth%4); //for checksuming
-
-            //give the catalog enough space
-            fCatalog.resize(fNumTiles);
-            for (uint32_t i=0;i<fNumTiles;i++)
-            {
-                fCatalog[i].resize(fRealColumns.size());
-                for (auto it=fCatalog[i].begin(); it!=fCatalog[i].end(); it++)
-                    *it = CatalogEntry(0,0);
-            }
         }
 
@@ -667 +656 @@
 
             fCheckOffset = (4 - extraBytes)%4;
+
             //write data to disk
             write(src+4, sizeToWrite);
@@ -679 +669 @@
         {
             uint64_t compressed_size = 0;
+
+            //Can't get this to work in the thread. Printed the adresses, and they seem to be correct.
+            //Really do not understand what's wrong...
+            //calibrate data if required
+            //const uint32_t thisRoundNumRows  = (target.num_rows%fNumRowsPerTile) ? target.num_rows%fNumRowsPerTile : fNumRowsPerTile;
+//            for (uint32_t i=0;i<thisRoundNumRows;i++)
+//            {
+//                char* target_location = target.src.get()->get() + fRealRowWidth*i;
+//                cout << "Target Location there...." << hex << static_cast<void*>(target_location) << endl;
+//                DrsOffsetCalibrate(target_location);
+//            }
+
 #ifdef __EXCEPTIONS
             try
@@ -687 +689 @@
 
                 //compress the buffer
-                compressed_size = compressBuffer(target.target.data.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, target.catalog_entry);
 #ifdef __EXCEPTIONS
             }
@@ -762 +764 @@
         /// @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)
+        uint64_t compressBuffer(char* dest, char* src, uint32_t num_rows, CatalogRow& catalog_row)
         {
             const uint32_t thisRoundNumRows  = (num_rows%fNumRowsPerTile) ? num_rows%fNumRowsPerTile : fNumRowsPerTile;
-            const uint32_t currentCatalogRow = (num_rows-1)/fNumRowsPerTile;
             uint32_t       offset            = 0;
 
@@ -777 +778 @@
             for (uint32_t i=0;i<fRealColumns.size();i++)
             {
-                fCatalog[currentCatalogRow][i].second = compressedOffset;
+                catalog_row[i].second = compressedOffset;
 
                 if (fRealColumns[i].col.num == 0) continue;
@@ -811 +812 @@
                 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;
                     //de-smooth !
                     if (head.getProc(0) == kFactSmoothing)
@@ -826 +826 @@
                     offset += thisRoundNumRows*fRealColumns[i].col.size*fRealColumns[i].col.num;
 
-                    fCatalog[currentCatalogRow][i].first = compressedOffset - fCatalog[currentCatalogRow][i].second;
+                    catalog_row[i].first = compressedOffset - catalog_row[i].second;
                     continue;
                 }
@@ -834 +834 @@
 
                 offset += thisRoundNumRows*fRealColumns[i].col.size*fRealColumns[i].col.num;
-                fCatalog[currentCatalogRow][i].first = compressedOffset - fCatalog[currentCatalogRow][i].second;
+                catalog_row[i].first = compressedOffset - catalog_row[i].second;
             }
 
@@ -968 +968 @@
 
         // catalog related stuff
-        struct CatalogEntry
-        {
-            CatalogEntry(int64_t f=0, int64_t s=0) : first(f), second(s) {};
-            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<CatalogEntry> CatalogRow;
-        typedef vector<CatalogRow>     CatalogType;
         CatalogType fCatalog;               ///< Catalog for this file
-//        uint32_t    fCatalogSize;          ///<  Actual catalog size (.size() is slow on large lists)
+        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    fCatalogExtraRows; ///<      Number of extra rows written on top of the initial capacity of the file
 
         // checksum related stuff