Changeset 17219

Timestamp:

10/15/13 15:05:54 (11 years ago)

Author:

lyard

Message:

draft zofits working with Queue and MemoryManager

Location:

trunk

Files:

• FACT++/src/queue.h (modified) (8 diffs)
• Mars/mcore/MemoryManager.h (added)
• Mars/mcore/Queue.h (added)
• Mars/mcore/zofits.h (modified) (26 diffs)

trunk/FACT++/src/queue.h

@@ -10 +10 @@
 {
     size_t fSize;                 // Only necessary for before C++11
+    bool   fSort;                 // Sort the list before processing
 
     std::list<T> fList;
 
-    std::mutex fMutex;        // Mutex needed for the conditional
+    std::mutex fMutex;             // Mutex needed for the conditional
     std::condition_variable fCond; // Conditional
 
@@ -22 +23 @@
         kStop,
         kAbort,
+        kTrigger
     };
 
     state_t fState;               // Stop signal for the thread
 
-    typedef std::function<void(const T &)> callback;
+    typedef std::function<bool(const T &)> callback;
     callback fCallback;       // Callback function called by the thread
 
@@ -34 +36 @@
     {
         std::unique_lock<std::mutex> lock(fMutex);
+
+        // No filling allowed by default (the queue is
+        // always processed until it is empty)
+        size_t allowed = 0;
 
         while (1)
         {
-            while (fList.empty() && fState==kRun)
+            while (fSize==allowed && fState==kRun)
                 fCond.wait(lock);
 
+            // Check if the State flag has been changed
             if (fState==kAbort)
                 break;
@@ -46 +53 @@
                 break;
 
-            const T &val = fList.front();
+            // If thread got just woken up, move back the state to kRun
+            if (fState == kTrigger)
+                fState = kRun;
+
+            // Could have been a fState==kTrigger case
+            if (fList.empty())
+                continue;
+
+            // During the unlocked state, fSize might change.
+            // The current size of the queue needs to be saved.
+            allowed = fSize;
+
+            // get the first entry from the (sorted) list
+            const auto it = fSort ? min_element(fList.begin(), fList.end()) : fList.begin();
 
             // Theoretically, we can loose a signal here, but this is
@@ -52 +72 @@
             lock.unlock();
 
-            if (fCallback)
-                fCallback(val);
+            // If the first event in the queue could not be processed,
+            // no further processing makes sense until a new event has
+            // been posted (or the number of events in the queue has
+            // changed)  [allowed>0], in the case processing was
+            // successfull [alloed==0], the next event will be processed
+            // immediately.
+            if (!fCallback || !fCallback(*it))
+                allowed = 0;
 
             lock.lock();
 
-            fList.pop_front();
+            // Whenever an event was successfully processed, allowed
+            // is larger than zero and thus the event will be popped
+            if (allowed==0)
+                continue;
+
+            if (fSort)
+                fList.erase(it);
+            else
+                fList.pop_front() ;
+
             fSize--;
+            allowed--;
+
         }
 
@@ -68 +105 @@
 
 public:
-    Queue(const callback &f) : fSize(0), fState(kIdle), fCallback(f)
-    {
-        start();
-    }
+    Queue(const callback &f, bool sort=false, bool startup=true) : fSize(0), fSort(sort), fState(kIdle), fCallback(f)
+    {
+        if (startup)
+            start();
+    }
+
+    Queue(const Queue<T>& q) : fSize(0), fSort(q.fSort), fState(kIdle), fCallback(q.fCallback)
+    {
+    }
+
+    Queue<T>& operator = (const Queue<T>& q)
+    {
+        fSize     = 0;
+        fSort     = q.fSort;
+        fState    = kIdle;
+        fCallback = q.fCallback;
+        return *this;
+    }
+
     ~Queue()
     {
@@ -133 +185 @@
     {
         const std::lock_guard<std::mutex> lock(fMutex);
+
         if (fState==kIdle)
             return false;
@@ -139 +192 @@
         fSize++;
 
+        fCond.notify_one();
+
+        return true;
+    }
+
+    bool notify()
+    {
+        const std::lock_guard<std::mutex> lock(fMutex);
+        if (fState!=kRun)
+            return false;
+
+        fState = kTrigger;
         fCond.notify_one();
 

trunk/Mars/mcore/zofits.h

@@ -7 +7 @@
 
 #include "ofits.h"
+#include "Queue.h"
+#include "MemoryManager.h"
 
 #ifndef __MARS__
@@ -64 +66 @@
 
 
+        struct WriteTarget
+        {
+            bool operator < (const WriteTarget& other)
+            {
+                tile_num < other.tile_num;
+            }
+            uint32_t tile_num;
+            uint32_t size;
+            shared_ptr<MemoryChunk> target;
+        };
+
+        struct CompressionTarget
+        {
+            bool operator < (const CompressionTarget& other)
+            {
+                return target < other.target;
+            }
+            shared_ptr<MemoryChunk> src;
+            WriteTarget             target;
+            uint32_t                num_rows;
+        };
+
+
         //constructors
         zofits(uint32_t numTiles=1000,
-               uint32_t rowPerTile=100) : ofits()
-        {
-            InitMemberVariables(numTiles, rowPerTile);
+               uint32_t rowPerTile=100,
+               uint64_t maxUsableMem=0) : ofits(),
+                                          fMemPool(0, maxUsableMem),
+                                          fWriteToDiskQueue(bind(&zofits::WriteBufferToDisk, this, placeholders::_1), true)
+        {
+            InitMemberVariables(numTiles, rowPerTile, maxUsableMem);
             SetNumWorkingThreads(1);
         }
@@ -74 +102 @@
         zofits(const char* fname,
                uint32_t numTiles=1000,
-               uint32_t rowPerTile=100) : ofits(fname)
-        {
-            InitMemberVariables(numTiles, rowPerTile);
+               uint32_t rowPerTile=100,
+               uint64_t maxUsableMem=0) : ofits(fname),
+                                          fMemPool(0, maxUsableMem),
+                                          fWriteToDiskQueue(bind(&zofits::WriteBufferToDisk, this, placeholders::_1), true)
+        {
+            InitMemberVariables(numTiles, rowPerTile, maxUsableMem);
             SetNumWorkingThreads(1);
         }
@@ -85 +116 @@
 
         //initialization of member variables
-        void InitMemberVariables(uint32_t nt=0, uint32_t rpt=0)
+        void InitMemberVariables(uint32_t nt=0, uint32_t rpt=0, uint64_t maxUsableMem=0)
         {
             fCheckOffset  = 0;
-            fThreadLooper = 0;
 
             fNumTiles       = nt;
             fNumRowsPerTile = rpt;
 
-            fNumThreads  = 1;
-            fThreadIndex = 0;   ///< A variable to assign threads indices
+            fNumQueues   = 0;
+            fQueueLooper = 0;
 
             fBuffer       = NULL;
@@ -102 +132 @@
             fStartCellsOffset = -1;
             fDataOffset       = -1;
+
+            fMaxUsableMem = maxUsableMem;
         }
 
@@ -144 +176 @@
         bool WriteTableHeader(const char* name="DATA")
         {
+            if (!reallocateBuffers())
+                throw ("While allocating memory: apparently there not as much free memory as advertized...");
+
             ofits::WriteTableHeader(name);
+
+            //start the compression queues
+            for (auto it=fCompressionQueues.begin(); it!= fCompressionQueues.end(); it++)
+                it->start();
+
+            //mark that no tile has been written so far
+            fLatestWrittenTile = -1;
 
             if (IsOffsetCalibrated())
@@ -208 +250 @@
 
             fRawSum.reset();
-
-            //start the compression threads
-            pthread_mutex_init(&fMutex, NULL);
-
-            fThreadIndex = 0;
-            for (uint32_t i=0;i<fNumThreads;i++)
-                pthread_create(&(fThread[i]), NULL, threadFunction, this);
-
-            //wait for all threads to have started
-            while (fNumThreads != fThreadIndex)
-                usleep(1000);
-
-            //set the writing fence to the last thread (so that the first one can start writing right away)
-            fThreadIndex = fNumThreads-1;
         }
 
@@ -391 +419 @@
 
             if (fTable.num_rows % fNumRowsPerTile == 0)
-            {//give a new tile to compress to a thread
-                while (fThreadStatus[fThreadLooper] == _THREAD_COMPRESS_)
-                    usleep(100000);
-
-                copyTransposeTile(fThreadLooper);
-
-                while (fThreadStatus[fThreadLooper] != _THREAD_WAIT_)
-                    usleep(100000);
-
-                fThreadNumRows[fThreadLooper] = fTable.num_rows;
-                fThreadStatus[fThreadLooper] = _THREAD_COMPRESS_;
-                fThreadLooper = (fThreadLooper+1)%fNumThreads;
+            {
+                CompressionTarget compress_target;
+                SetNextCompression(compress_target);
+
+                if (!fCompressionQueues[fQueueLooper].post(compress_target))
+                    throw runtime_error("I could not post this buffer. This does not make sense...");
+
+                fQueueLooper = (fQueueLooper+1)%fNumQueues;
             }
 
@@ -415 +439 @@
         }
 
+        void SetNextCompression(CompressionTarget& target)
+        {
+            shared_ptr<MemoryChunk> transposed_data = fMemPool.malloc();
+
+            copyTransposeTile(fBuffer, transposed_data.get()->get());
+
+            WriteTarget write_target;
+            write_target.tile_num = (fTable.num_rows-1)/fNumRowsPerTile;
+            write_target.size     = 0;
+            write_target.target   = fMemPool.malloc();
+
+            target.src      = transposed_data;
+            target.target   = write_target;
+            target.num_rows = fTable.num_rows;
+        }
+
         bool close()
         {
@@ -420 +460 @@
                 return false;
 
-            //wait for compression threads to finish
-            for (auto it=fThreadStatus.begin(); it!= fThreadStatus.end(); it++)
-                while (*it != _THREAD_WAIT_)
-                    usleep(100000);
-
-            for (auto it=fThreadStatus.begin(); it!= fThreadStatus.end(); it++)
-                *it = _THREAD_EXIT_;
-
-            for (auto it=fThread.begin(); it!= fThread.end(); it++)
-                pthread_join(*it, NULL);
-
-            pthread_mutex_destroy(&fMutex);
+            for (auto it=fCompressionQueues.begin(); it != fCompressionQueues.end(); it++)
+                it->wait();
+
+            fWriteToDiskQueue.wait();
 
             if (fTable.num_rows%fNumRowsPerTile != 0)
             {
-                copyTransposeTile(0);
-                fThreadNumRows[0] = fTable.num_rows;
-                uint32_t numBytes = compressBuffer(0);
-                writeCompressedDataToDisk(0, numBytes);
+                CompressionTarget compress_target;
+                SetNextCompression(compress_target);
+
+                uint64_t size_to_write = CompressBuffer(compress_target);
+
+                WriteTarget write_target;
+                write_target.size     = size_to_write;
+                write_target.target   = compress_target.target.target;
+                write_target.tile_num = compress_target.target.tile_num;
+
+                if (!WriteBufferToDisk(write_target))
+                    throw runtime_error("Something went wrong while writing the last tile...");
             }
 
@@ -449 +489 @@
             uint64_t heap_offset = fCatalog.size()*fTable.num_cols*sizeof(uint64_t)*2;
             SetInt("ZHEAPPTR", heap_offset);
-//            SetInt("THEAP"   , heap_offset);0
 
             const uint32_t total_num_tiles_written = (fTable.num_rows + fNumRowsPerTile-1)/fNumRowsPerTile;
@@ -547 +586 @@
             SetStr(strKey.str(), strVal.str(), strCom.str());
 
-            return reallocateBuffers();
+            return true;
         }
 
@@ -571 +610 @@
             }
 
-            fNumThreads = num;
-            fThreadStatus.resize(num);
-            fThread.resize(num);
-            fThreadNumRows.resize(num);
-            for (uint32_t i=0;i<num;i++)
-            {
-                fThreadNumRows[i] = 0;
-                fThreadStatus[i] = _THREAD_WAIT_;
-            }
-
-            return reallocateBuffers();
-        }
+            if (fCompressionQueues.size() == num)
+                return true;
+
+            //cannot be const, as resize does not want it that way
+            Queue<CompressionTarget> queue(bind(&zofits::CompressBuffer, this, placeholders::_1), false, false);
+
+            //shrink
+            if (num < fCompressionQueues.size())
+            {
+                fCompressionQueues.resize(num, queue);
+                return true;
+            }
+
+            //grow
+            fCompressionQueues.resize(num, queue);
+
+            fNumQueues   = num;
+            fQueueLooper = 0;
+
+            return true;
+        }
+
 
     private:
 
-
         bool reallocateBuffers()
         {
-            fBufferVector.resize(fRealRowWidth*fNumRowsPerTile + 8); //8 more for checksuming
-            memset(fBufferVector.data(), 0, 4);
-            fBuffer = fBufferVector.data()+4;
+            size_t chunk_size = fRealRowWidth*fNumRowsPerTile + fRealColumns.size()*sizeof(BlockHeader) + sizeof(TileHeader) + 8; //+8 for checksuming;
+            fMemPool.setChunkSize(chunk_size);
+
+            fSmartBuffer = fMemPool.malloc();
+            fBuffer = fSmartBuffer.get()->get();
+//            memset(fBuffer, 0, 4);
+//            fBuffer += 4;
 
             fRawSumBuffer.resize(fRealRowWidth + 4-fRealRowWidth%4); //for checksuming
-
-
-            fTransposedBufferVector.resize(fNumThreads);
-            fCompressedBufferVector.resize(fNumThreads);
-            fTransposedBuffer.resize(fNumThreads);
-            fCompressedBuffer.resize(fNumThreads);
-            for (uint32_t i=0;i<fNumThreads;i++)
-            {
-                fTransposedBufferVector[i].resize(fRealRowWidth*fNumRowsPerTile);
-                fCompressedBufferVector[i].resize(fRealRowWidth*fNumRowsPerTile + fRealColumns.size() + sizeof(TileHeader) + 8);
-                memset(fCompressedBufferVector[i].data(), 0, 4);
-                TileHeader tileHeader;
-                memcpy(fCompressedBufferVector[i].data()+4, &tileHeader, sizeof(TileHeader));
-                fTransposedBuffer[i] = fTransposedBufferVector[i].data();
-                fCompressedBuffer[i] = fCompressedBufferVector[i].data()+4;
-            }
 
             //give the catalog enough space
@@ -622 +658 @@
         }
 
-        bool writeCompressedDataToDisk(uint32_t threadID, uint32_t sizeToWrite)
-        {
-            char* checkSumPointer = fCompressedBuffer[threadID];
+        bool writeCompressedDataToDisk(char* src, uint32_t sizeToWrite)
+        {
+            char* checkSumPointer = src+4;
             int32_t extraBytes = 0;
             uint32_t sizeToChecksum = sizeToWrite;
@@ -645 +681 @@
             fCheckOffset = (4 - extraBytes)%4;
             //write data to disk
-            write(fCompressedBuffer[threadID], sizeToWrite);
+            write(src+4, sizeToWrite);
+
             return good();
         }
 
-        static void* threadFunction(void* context)
-        {
-            zofits* myself = static_cast<zofits*>(context);
-
-            uint32_t myID = 0;
-            pthread_mutex_lock(&(myself->fMutex));
-            myID = myself->fThreadIndex++;
-            pthread_mutex_unlock(&(myself->fMutex));
-            uint32_t threadToWaitForBeforeWriting = (myID == 0) ? myself->fNumThreads-1 : myID-1;
-
-            while (myself->fThreadStatus[myID] != _THREAD_EXIT_)
-            {
-                while (myself->fThreadStatus[myID] == _THREAD_WAIT_)
-                    usleep(100000);
-
-                if (myself->fThreadStatus[myID] != _THREAD_COMPRESS_)
-                    continue;
-                uint32_t numBytes = myself->compressBuffer(myID);
-                myself->fThreadStatus[myID] = _THREAD_WRITE_;
-
-                //wait for the previous data to be written
-                while (myself->fThreadIndex != threadToWaitForBeforeWriting)
-                    usleep(1000);
-                //do the actual writing to disk
-                pthread_mutex_lock(&(myself->fMutex));
-                myself->writeCompressedDataToDisk(myID, numBytes);
-                myself->fThreadIndex = myID;
-                pthread_mutex_unlock(&(myself->fMutex));
-                myself->fThreadStatus[myID] = _THREAD_WAIT_;
-            }
-            return NULL;
-        }
-
-        uint64_t compressBuffer(uint32_t threadIndex)
-        {
-            uint32_t thisRoundNumRows = (fThreadNumRows[threadIndex]%fNumRowsPerTile) ? fThreadNumRows[threadIndex]%fNumRowsPerTile : fNumRowsPerTile;
+        bool CompressBuffer(const CompressionTarget& target)
+        {
+            //compress the buffer
+            uint64_t compressed_size = compressBuffer(target.target.target.get()->get(), target.src.get()->get(), target.num_rows);
+
+            //post the result to the writing queue
+            //get a copy so that it becomes non-const
+            WriteTarget wt;
+            wt.tile_num = target.target.tile_num;
+            wt.size     = compressed_size;
+            wt.target   = target.target.target;
+
+            fWriteToDiskQueue.post(wt);
+            return true;
+        }
+
+        bool WriteBufferToDisk(const WriteTarget& target)
+        {
+            //is this the tile we're supposed to write ?
+            if (target.tile_num != fLatestWrittenTile+1)
+                return false;
+
+            fLatestWrittenTile++;
+
+            //write the buffer to disk.
+            writeCompressedDataToDisk(target.target.get()->get(), target.size);
+
+            return true;
+        }
+
+        //src cannot be const, as applySMOOTHING is done in place
+        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 = (fThreadNumRows[threadIndex]-1)/fNumRowsPerTile;
-            uint64_t compressedOffset = sizeof(TileHeader); //skip the 'TILE' marker and tile size entry
+            uint32_t currentCatalogRow = (num_rows-1)/fNumRowsPerTile;
+
+            //skip the checksum reserved area
+            dest += 4;
+
+            //skip the 'TILE' marker and tile size entry
+            uint64_t compressedOffset = sizeof(TileHeader);
 
             //now compress each column one by one by calling compression on arrays
@@ -709 +749 @@
                     {
                         case zfits::kFactRaw:
-                                compressedOffset += compressUNCOMPRESSED(&(fCompressedBuffer[threadIndex][compressedOffset]),
-                                                                         &(fTransposedBuffer[threadIndex][offset]),
+                                compressedOffset += compressUNCOMPRESSED(dest + compressedOffset,
+                                                                         src  + offset,
                                                                          thisRoundNumRows,
                                                                          fRealColumns[i].col.size,
@@ -716 +756 @@
                         break;
                         case zfits::kFactSmoothing:
-                                applySMOOTHING(&(fCompressedBuffer[threadIndex][compressedOffset]),
-                                               &(fTransposedBuffer[threadIndex][offset]),
+                                applySMOOTHING(dest + compressedOffset,
+                                               src  + offset,
                                                thisRoundNumRows,
                                                fRealColumns[i].col.size,
@@ -724 +764 @@
                         case zfits::kFactHuffman16:
                             if (head.ordering == zfits::kOrderByCol)
-                                compressedOffset += compressHUFFMAN(&(fCompressedBuffer[threadIndex][compressedOffset]),
-                                                                    &(fTransposedBuffer[threadIndex][offset]),
+                                compressedOffset += compressHUFFMAN(dest + compressedOffset,
+                                                                    src  + offset,
                                                                     thisRoundNumRows,
                                                                     fRealColumns[i].col.size,
                                                                     fRealColumns[i].col.num);
                             else
-                                compressedOffset += compressHUFFMAN(&(fCompressedBuffer[threadIndex][compressedOffset]),
-                                                                    &(fTransposedBuffer[threadIndex][offset]),
+                                compressedOffset += compressHUFFMAN(dest + compressedOffset,
+                                                                    src  + offset,
                                                                     fRealColumns[i].col.num,
                                                                     fRealColumns[i].col.size,
@@ -746 +786 @@
                     compressedOffset - previousOffset > fRealColumns[i].col.size*fRealColumns[i].col.num*thisRoundNumRows+sizeof(BlockHeader)+sizeof(uint16_t)*sequence.size())
                 {//if so set flag and redo it uncompressed
-                    //cout << "REDOING UNCOMPRESSED" << endl;
+                    cout << "REDOING UNCOMPRESSED" << endl;
                     compressedOffset = previousOffset + sizeof(BlockHeader) + 1;
-                    compressedOffset += compressUNCOMPRESSED(&(fCompressedBuffer[threadIndex][compressedOffset]), &(fTransposedBuffer[threadIndex][offset]), thisRoundNumRows, fRealColumns[i].col.size, fRealColumns[i].col.num);
+                    compressedOffset += compressUNCOMPRESSED(dest + compressedOffset, src + offset, thisRoundNumRows, fRealColumns[i].col.size, fRealColumns[i].col.num);
                     BlockHeader he;
                     he.size = compressedOffset - previousOffset;
                     he.numProcs = 1;
                     he.ordering = zfits::kOrderByRow;
-                    memcpy(&(fCompressedBuffer[threadIndex][previousOffset]), (char*)(&he), sizeof(BlockHeader));
-                    fCompressedBuffer[threadIndex][previousOffset+sizeof(BlockHeader)] = zfits::kFactRaw;
+                    memcpy(dest + previousOffset, (char*)(&he), sizeof(BlockHeader));
+                    dest[previousOffset+sizeof(BlockHeader)] = zfits::kFactRaw;
                     offset += thisRoundNumRows*fRealColumns[i].col.size*fRealColumns[i].col.num;
-                   fCatalog[currentCatalogRow][i].first = compressedOffset - fCatalog[currentCatalogRow][i].second;
-                   continue;
+                    fCatalog[currentCatalogRow][i].first = compressedOffset - fCatalog[currentCatalogRow][i].second;
+                    continue;
                 }
+
                 head.size = compressedOffset - previousOffset;
-                memcpy(&(fCompressedBuffer[threadIndex][previousOffset]), (char*)(&head), sizeof(BlockHeader));
-                memcpy(&(fCompressedBuffer[threadIndex][previousOffset+sizeof(BlockHeader)]), sequence.data(), sizeof(uint16_t)*sequence.size());
-
-                 offset += thisRoundNumRows*fRealColumns[i].col.size*fRealColumns[i].col.num;
+                memcpy(dest + previousOffset, (char*)(&head), sizeof(BlockHeader));
+                memcpy(dest + previousOffset+sizeof(BlockHeader), sequence.data(), sizeof(uint16_t)*sequence.size());
+
+                offset += thisRoundNumRows*fRealColumns[i].col.size*fRealColumns[i].col.num;
                 fCatalog[currentCatalogRow][i].first = compressedOffset - fCatalog[currentCatalogRow][i].second;
             }
 
-            TileHeader tHead(thisRoundNumRows, compressedOffset);
-            memcpy(fCompressedBuffer[threadIndex], &tHead, sizeof(TileHeader));
+            TileHeader tile_head(thisRoundNumRows, compressedOffset);
+            memcpy(dest, &tile_head, sizeof(TileHeader));
+
             return compressedOffset;
         }
 
-        void copyTransposeTile(uint32_t index)
+        void copyTransposeTile(const char* src, char* dest)//uint32_t index)
         {
             uint32_t thisRoundNumRows = (fTable.num_rows%fNumRowsPerTile) ? fTable.num_rows%fNumRowsPerTile : fNumRowsPerTile;
 
             //copy the tile and transpose it
-            uint32_t offset = 0;
             for (uint32_t i=0;i<fRealColumns.size();i++)
             {
@@ -785 +826 @@
                         for (uint32_t k=0;k<thisRoundNumRows;k++)
                         {//regular, "semi-transposed" copy
-                            memcpy(&(fTransposedBuffer[index][offset]), &fBuffer[k*fRealRowWidth + fRealColumns[i].col.offset], fRealColumns[i].col.size*fRealColumns[i].col.num);
-                            offset += fRealColumns[i].col.size*fRealColumns[i].col.num;
+                            memcpy(dest, src+k*fRealRowWidth+fRealColumns[i].col.offset, fRealColumns[i].col.size*fRealColumns[i].col.num);
+                            dest += fRealColumns[i].col.size*fRealColumns[i].col.num;
                         }
                     break;
@@ -794 +835 @@
                             for (uint32_t k=0;k<thisRoundNumRows;k++)
                             {//transposed copy
-                                memcpy(&(fTransposedBuffer[index][offset]), &fBuffer[k*fRealRowWidth + fRealColumns[i].col.offset + fRealColumns[i].col.size*j], fRealColumns[i].col.size);
-                                offset += fRealColumns[i].col.size;
+                                memcpy(dest, src+k*fRealRowWidth+fRealColumns[i].col.offset+fRealColumns[i].col.size*j, fRealColumns[i].col.size);
+                                dest += fRealColumns[i].col.size;
                             }
                     break;
                     default:
                             cout << "Error: unknown column ordering: " << fRealColumns[i].head.ordering << endl;
-
                 };
             }
@@ -844 +884 @@
         }
 
-        uint32_t compressSMOOTHMAN(char* dest, 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;
-            //call the huffman transposed
-            return compressHUFFMAN(dest, src, numRowElems, sizeOfElems, numRows);
-        }
-
         uint32_t applySMOOTHING(char* dest, char* src, uint32_t numRows, uint32_t sizeOfElems, uint32_t numRowElems)
         {
@@ -873 +904 @@
         uint32_t        fNumRowsPerTile;
 
-
         //thread related stuff
-        uint32_t          fNumThreads;    ///< The number of threads that will be used to compress
-        uint32_t          fThreadIndex;   ///< A variable to assign threads indices
-        vector<pthread_t> fThread;        ///< The thread handler of the compressor
-        vector<uint32_t>  fThreadNumRows; ///< Total number of rows for thread to compresswww.;wwwwww
-        vector<uint32_t>  fThreadStatus;  ///< Flag telling whether the buffer to be transposed (and compressed) is full or empty
-        int32_t           fThreadLooper;      ///< Which thread will deal with the upcoming bunch of data ?
-        pthread_mutex_t   fMutex;             ///< mutex for compressing threads
+        vector<Queue<CompressionTarget>> fCompressionQueues;
+        Queue<WriteTarget>               fWriteToDiskQueue;
+
+        //thread related stuff
+        uint32_t          fNumQueues;    ///< The number of threads that will be used to compress
+        uint32_t          fQueueLooper;
+        int32_t           fLatestWrittenTile;
 
         struct CatalogEntry
@@ -890 +920 @@
         } __attribute__((__packed__));
 
-       // typedef pair<int64_t, int64_t> CatalogEntry;
         typedef vector<CatalogEntry>   CatalogRow;
         typedef vector<CatalogRow>     CatalogType;
@@ -897 +926 @@
         Checksum             fRawSum;
         off_t                fCatalogOffset;
-        vector<char>         fBufferVector;
+        uint32_t             fRealRowWidth;
+
         vector<char>         fRawSumBuffer;
-        vector<vector<char>> fTransposedBufferVector;
-        vector<vector<char>> fCompressedBufferVector;
-        char*                fBuffer;
-        vector<char*>        fTransposedBuffer;
-        vector<char*>        fCompressedBuffer;
-        uint32_t             fRealRowWidth;
+        MemoryManager        fMemPool;
+        uint64_t             fMaxUsableMem;
+
+        shared_ptr<MemoryChunk> fSmartBuffer;
+        char*                   fBuffer;
+
+
         struct CompressedColumn
         {
@@ -917 +948 @@
         vector<CompressedColumn> fRealColumns;
 
-        //thread states. Not all used, but they do not hurt
-        static const uint32_t       _THREAD_WAIT_ = 0; ///< Thread doing nothing
-        static const uint32_t   _THREAD_COMPRESS_ = 1; ///< Thread working, compressing
-        static const uint32_t _THREAD_DECOMPRESS_ = 2; ///< Thread working, decompressing
-        static const uint32_t      _THREAD_WRITE_ = 3; ///< Thread writing data to disk
-        static const uint32_t       _THREAD_READ_ = 4; ///< Thread reading data from disk
-        static const uint32_t       _THREAD_EXIT_ = 5; ///< Thread exiting
-
 };