Changeset 17254

Timestamp:

10/18/13 14:10:24 (11 years ago)

Author:

tbretz

Message:

Use a shared_ptr with aliasing of char* instead of a MemoryChunk

Location:

trunk/Mars/mcore

Files:

• MemoryManager.h (modified) (4 diffs)
• zofits.h (modified) (14 diffs)

trunk/Mars/mcore/MemoryManager.h

@@ -99 +99 @@
 
     std::shared_ptr<MemoryStock> fMemoryStock;
-    std::shared_ptr<char>   fPointer;
-    char*                   fRealPointer;
+    std::shared_ptr<char>        fPointer;
 
-    MemoryChunk(const std::shared_ptr<MemoryStock> &mem, bool block) : fMemoryStock(mem)
+    MemoryChunk(const std::shared_ptr<MemoryStock> &mem, bool block)
+        : fMemoryStock(mem)
     {
         fPointer = fMemoryStock->pop(block);
-        fRealPointer = fPointer.get();
     }
 
@@ -113 +112 @@
         fMemoryStock->push(fPointer);
     }
-
-public:
-    char* get() { return fPointer.get();}
 };
 
@@ -123 +119 @@
 
 public:
-
     MemoryManager(size_t chunk, size_t max) : fMemoryStock(std::make_shared<MemoryStock>(chunk, max))
     {
     }
 
-    std::shared_ptr<MemoryChunk> malloc(bool block=true)
+    std::shared_ptr<char> malloc(bool block=true)
     {
-        MemoryChunk *chunk = new MemoryChunk(fMemoryStock, block);
-        //Etienne cannot get the aliasing to work (at least with g++ 4.4.6
-        return std::shared_ptr<MemoryChunk>(std::shared_ptr<MemoryChunk>(chunk));//, chunk->fRealPointer);
+        const std::shared_ptr<MemoryChunk> chunk(new MemoryChunk(fMemoryStock, block));
+        return std::shared_ptr<char>(chunk, chunk->fPointer.get());
     }
 
@@ -142 +136 @@
             throw std::runtime_error("Cannot change the chunk size while there is memory in use");
         if (getMaxMemory()<size)
-            throw std::runtime_error("Chunk size("+to_string(size)+") larger thann allowed memory ("+to_string(getMaxMemory())+")");
+            throw std::runtime_error("Chunk size ("+std::to_string(size)+") larger than allowed memory ("+std::to_string(getMaxMemory())+")");
 #else
         if (getInUse() || getMaxMemory()<size)

trunk/Mars/mcore/zofits.h

@@ -37 +37 @@
         };
 
+
         //catalog types
         struct CatalogEntry
@@ -42 +43 @@
             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
+            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
@@ -55 +58 @@
             }
 
-            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<MemoryChunk> data;   ///<   Memory block to write
+            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<char> data;     ///<  Memory block to write
         };
+
 
         /// Parameters required to compress a tile of data
@@ -66 +70 @@
             {}
 
-            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
+            CatalogRow&      catalog_entry;   ///< Reference to the catalog entry to deal with
+            shared_ptr<char> src;             ///< Original data
+            shared_ptr<char> transposed_src;  ///< Transposed data
+            WriteTarget      target;          ///< Compressed data
+            uint32_t         num_rows;        ///< Number of rows to compress
          };
 
@@ -124 +128 @@
             fNumRowsPerTile = rpt;
 
-            fBuffer           = NULL;
             fRealRowWidth     = 0;
             fCatalogOffset    = 0;
@@ -273 +276 @@
 
             //copy current row to pool or rows waiting for compression
-            char* target_location = fBuffer + fRealRowWidth*(fTable.num_rows%fNumRowsPerTile);
+            char* target_location = fSmartBuffer.get() + fRealRowWidth*(fTable.num_rows%fNumRowsPerTile);
             memcpy(target_location, ptr, fRealRowWidth);
 
@@ -309 +312 @@
                 if (fNumQueues == 0)
                 { //no worker threads. do everything in-line
-                    uint64_t size_to_write = CompressBuffer(compress_target);
+                    const uint64_t size_to_write = CompressBuffer(compress_target);
 
                     WriteTarget write_target;
@@ -325 +328 @@
                      uint32_t current_index = 0;
 
-                     for (auto it=fCompressionQueues.begin(); it!=fCompressionQueues.end(); it++)
+                     for (auto it=fCompressionQueues.cbegin(); it!=fCompressionQueues.cend(); it++)
                      {
                          if (it->size() < min_size)
@@ -356 +359 @@
         {
             //get space for transposed data
-            shared_ptr<MemoryChunk> transposed_data = fMemPool.malloc();
+            shared_ptr<char> transposed_data = fMemPool.malloc();
 
             //fill up write to disk target
@@ -372 +375 @@
             //get a new buffer to host the incoming data
             fSmartBuffer = fMemPool.malloc();
-            fBuffer      = fSmartBuffer.get()->get();
         }
 
@@ -625 +627 @@
 
             fSmartBuffer = fMemPool.malloc();
-            fBuffer      = fSmartBuffer.get()->get();
-
             fRawSumBuffer.resize(fRealRowWidth + 4-fRealRowWidth%4); //for checksuming
         }
@@ -686 +686 @@
 #endif
                 //transpose the original data
-                copyTransposeTile(target.src.get()->get(), target.transposed_src.get()->get());
+                copyTransposeTile(target.src.get(), target.transposed_src.get());
 
                 //compress the buffer
-                compressed_size = compressBuffer(target.target.data.get()->get(), target.transposed_src.get()->get(), target.num_rows, target.catalog_entry);
+                compressed_size = compressBuffer(target.target.data.get(), target.transposed_src.get(), target.num_rows, target.catalog_entry);
 #ifdef __EXCEPTIONS
             }
@@ -730 +730 @@
             {
 #endif
-                if (!writeCompressedDataToDisk(target.data.get()->get(), target.size))
+                if (!writeCompressedDataToDisk(target.data.get(), target.size))
                 {//could not write the data to disk
                     ostringstream str;
@@ -990 +990 @@
         };
         vector<CompressedColumn> fRealColumns;     ///< Vector hosting the columns of the file
-        uint32_t                 fRealRowWidth;   ///<  Width in bytes of one uncompressed row
-        shared_ptr<MemoryChunk>  fSmartBuffer;   ///<   Smart pointer to the buffer where the incoming rows are written
-        char*                    fBuffer;       ///<    regular version of fSmartBuffer
-        vector<char>             fRawSumBuffer;///<     buffer used for checksuming the incoming data, before compression
+        uint32_t                 fRealRowWidth;    ///< Width in bytes of one uncompressed row
+        shared_ptr<char>         fSmartBuffer;     ///< Smart pointer to the buffer where the incoming rows are written
+        vector<char>             fRawSumBuffer;    ///< buffer used for checksuming the incoming data, before compression
 
 #ifdef __EXCEPTIONS