Index: trunk/Mars/mcore/MemoryManager.h
===================================================================
--- trunk/Mars/mcore/MemoryManager.h	(revision 17219)
+++ trunk/Mars/mcore/MemoryManager.h	(revision 17219)
@@ -0,0 +1,159 @@
+#include <forward_list>
+
+class MemoryStock
+{
+    friend class MemoryChunk;
+    friend class MemoryManager;
+
+    size_t fChunkSize;
+    size_t fMaxMemory;
+
+    size_t fInUse;
+    size_t fAllocated;
+
+    size_t fMaxInUse;
+
+    std::mutex fMutexMem;
+    std::mutex fMutexCond;
+    std::condition_variable fCond;
+
+    std::forward_list<std::shared_ptr<char>> fMemoryStock;
+
+public:
+    MemoryStock(size_t chunk, size_t max) : fChunkSize(chunk), fMaxMemory(max),
+        fInUse(0), fAllocated(0), fMaxInUse(0)
+    {
+        if (chunk>max)
+            throw std::runtime_error("Size mismatch: Size of a single chunk exceeds maximum size of memory");
+    }
+
+private:
+    std::shared_ptr<char> pop(bool block)
+    {
+        if (block)
+        {
+            // No free slot available, next alloc would exceed max memory:
+            // block until a slot is available
+            std::unique_lock<std::mutex> lock(fMutexCond);
+            while (fMemoryStock.empty() && fAllocated+fChunkSize>fMaxMemory)
+                fCond.wait(lock);
+        }
+        else
+        {
+            // No free slot available, next alloc would exceed max memory
+            // return an empty pointer
+            if (fMemoryStock.empty() && fAllocated+fChunkSize>fMaxMemory)
+                return std::shared_ptr<char>();
+        }
+
+        // We will return this amount of memory
+        // This is not 100% thread safe, but it is not a super accurate measure anyway
+        fInUse += fChunkSize;
+        if (fInUse>fMaxInUse)
+            fMaxInUse = fInUse;
+
+        if (fMemoryStock.empty())
+        {
+            // No free slot available, allocate a new one
+            fAllocated += fChunkSize;
+            return std::shared_ptr<char>(new char[fChunkSize]);
+        }
+
+        // Get the next free slot from the stack and return it
+        const std::lock_guard<std::mutex> lock(fMutexMem);
+
+        const auto mem = fMemoryStock.front();
+        fMemoryStock.pop_front();
+        return mem;
+    };
+
+    void push(const std::shared_ptr<char> &mem)
+    {
+        if (!mem)
+            return;
+
+        // Decrease the amont of memory in use accordingly
+        fInUse -= fChunkSize;
+
+        // If the maximum memory has changed, we might be over the limit.
+        // In this case: free a slot
+        if (fAllocated>fMaxMemory)
+        {
+            fAllocated -= fChunkSize;
+            return;
+        }
+
+        {
+            const std::lock_guard<std::mutex> lock(fMutexMem);
+            fMemoryStock.emplace_front(mem);
+        }
+
+        {
+            const std::lock_guard<std::mutex> lock(fMutexCond);
+            fCond.notify_one();
+        }
+    }
+};
+
+class MemoryChunk
+{
+    friend class MemoryManager;
+
+    std::shared_ptr<MemoryStock> fMemoryStock;
+    std::shared_ptr<char>   fPointer;
+    char*                   fRealPointer;
+
+    MemoryChunk(const std::shared_ptr<MemoryStock> &mem, bool block) : fMemoryStock(mem)
+    {
+        fPointer = fMemoryStock->pop(block);
+        fRealPointer = fPointer.get();
+    }
+
+public:
+    ~MemoryChunk()
+    {
+        fMemoryStock->push(fPointer);
+    }
+
+public:
+    char* get() { return fPointer.get();}
+};
+
+class MemoryManager
+{
+    std::shared_ptr<MemoryStock> fMemoryStock;
+
+public:
+
+    MemoryManager(size_t chunk, size_t max) : fMemoryStock(std::make_shared<MemoryStock>(chunk, max))
+    {
+    }
+
+    std::shared_ptr<MemoryChunk> 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);
+    }
+
+    size_t getChunkSize() const { return fMemoryStock->fChunkSize; }
+    void   setChunkSize(const size_t size)
+    {
+        if (getInUse() != 0)
+            throw std::runtime_error("Cannot change the chunk size while there is memory in use");
+
+        if (getMaxMemory() < size)
+        {
+            std::ostringstream str;
+            str << "Chunk size(" << size << ") larger thank allowed memory(" << getMaxMemory() << "). Cannot allocate a single chunk.";
+            throw std::runtime_error(str.str());
+        }
+        fMemoryStock->fChunkSize = size;
+    }
+    size_t getMaxMemory() const { return fMemoryStock->fMaxMemory; }
+    size_t getInUse() const { return fMemoryStock->fInUse; }
+    size_t getAllocated() const { return fMemoryStock->fAllocated; }
+    size_t getMaxInUse() const { return fMemoryStock->fMaxInUse; }
+};
+
+
Index: trunk/Mars/mcore/Queue.h
===================================================================
--- trunk/Mars/mcore/Queue.h	(revision 17219)
+++ trunk/Mars/mcore/Queue.h	(revision 17219)
@@ -0,0 +1,263 @@
+#ifndef FACT_Queue
+#define FACT_Queue
+
+#include <list>
+#include <thread>
+#include <condition_variable>
+
+template<class T>
+class Queue
+{
+    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::condition_variable fCond; // Conditional
+
+    enum state_t
+    {
+        kIdle,
+        kRun,
+        kStop,
+        kAbort,
+        kTrigger
+    };
+
+    state_t fState;               // Stop signal for the thread
+
+    typedef std::function<bool(const T &)> callback;
+    callback fCallback;       // Callback function called by the thread
+
+    std::thread fThread;      // Handle to the thread
+
+    void Thread()
+    {
+        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 (fSize==allowed && fState==kRun)
+                fCond.wait(lock);
+
+            // Check if the State flag has been changed
+            if (fState==kAbort)
+                break;
+
+            if (fState==kStop && fList.empty())
+                break;
+
+            // 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
+            // not a problem, because then we detect a non-empty queue
+            lock.unlock();
+
+            // 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();
+
+            // 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--;
+
+        }
+
+        fList.clear();
+        fSize = 0;
+
+        fState = kIdle;
+    }
+
+public:
+    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()
+    {
+        wait(true);
+    }
+
+    bool start()
+    {
+        const std::lock_guard<std::mutex> lock(fMutex);
+        if (fState!=kIdle)
+            return false;
+
+        fState = kRun;
+        fThread = std::thread(std::bind(&Queue::Thread, this));
+        return true;
+    }
+
+    bool stop()
+    {
+        const std::lock_guard<std::mutex> lock(fMutex);
+        if (fState==kIdle)
+            return false;
+
+        fState = kStop;
+        fCond.notify_one();
+
+        return true;
+    }
+
+    bool abort()
+    {
+        const std::lock_guard<std::mutex> lock(fMutex);
+        if (fState==kIdle)
+            return false;
+
+        fState = kAbort;
+        fCond.notify_one();
+
+        return true;
+    }
+
+    bool wait(bool abrt=false)
+    {
+        {
+            const std::lock_guard<std::mutex> lock(fMutex);
+            if (fState==kIdle)
+                return false;
+
+            if (fState==kRun)
+            {
+                fState = abrt ? kAbort : kStop;
+                fCond.notify_one();
+            }
+        }
+
+        fThread.join();
+        return true;
+    }
+
+    bool post(const T &val)
+    {
+        const std::lock_guard<std::mutex> lock(fMutex);
+
+        if (fState==kIdle)
+            return false;
+
+        fList.push_back(val);
+        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();
+
+        return true;
+    }
+
+#ifdef __GXX_EXPERIMENTAL_CXX0X__
+    template<typename... _Args>
+        bool emplace(_Args&&... __args)
+    {
+        const std::lock_guard<std::mutex> lock(fMutex);
+        if (fState==kIdle)
+            return false;
+
+        fList.emplace_back(__args...);
+        fSize++;
+
+        fCond.notify_one();
+
+        return true;
+    }
+
+    bool post(T &&val) { return emplace(std::move(val)); }
+#endif
+
+#ifdef __GXX_EXPERIMENTAL_CXX0X__
+    bool move(std::list<T>&& x, typename std::list<T>::iterator i)
+#else
+    bool move(std::list<T>& x, typename std::list<T>::iterator i)
+#endif
+    {
+        const std::lock_guard<std::mutex> lock(fMutex);
+        if (fState==kIdle)
+            return false;
+
+        fList.splice(fList.end(), x, i);
+        fSize++;
+
+        fCond.notify_one();
+
+        return true;
+    }
+
+#ifdef __GXX_EXPERIMENTAL_CXX0X__
+    bool move(std::list<T>& x, typename std::list<T>::iterator i) { return move(std::move(x), i); }
+#endif
+
+    size_t size() const
+    {
+        return fSize;
+    }
+
+    bool empty() const
+    {
+        return fSize==0;
+    }
+};
+
+#endif
Index: trunk/Mars/mcore/zofits.h
===================================================================
--- trunk/Mars/mcore/zofits.h	(revision 17216)
+++ trunk/Mars/mcore/zofits.h	(revision 17219)
@@ -7,4 +7,6 @@
 
 #include "ofits.h"
+#include "Queue.h"
+#include "MemoryManager.h"
 
 #ifndef __MARS__
@@ -64,9 +66,35 @@
 
 
+        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,7 +102,10 @@
         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,14 +116,13 @@
 
         //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,4 +132,6 @@
             fStartCellsOffset = -1;
             fDataOffset       = -1;
+
+            fMaxUsableMem = maxUsableMem;
         }
 
@@ -144,5 +176,15 @@
         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,18 +250,4 @@
 
             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,16 +419,12 @@
 
             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,4 +439,20 @@
         }
 
+        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,23 +460,23 @@
                 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,5 +489,4 @@
             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,5 +586,5 @@
             SetStr(strKey.str(), strVal.str(), strCom.str());
 
-            return reallocateBuffers();
+            return true;
         }
 
@@ -571,43 +610,40 @@
             }
 
-            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,7 +658,7 @@
         }
 
-        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,47 +681,51 @@
             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,6 +749,6 @@
                     {
                         case zfits::kFactRaw:
-                                compressedOffset += compressUNCOMPRESSED(&(fCompressedBuffer[threadIndex][compressedOffset]),
-                                                                         &(fTransposedBuffer[threadIndex][offset]),
+                                compressedOffset += compressUNCOMPRESSED(dest + compressedOffset,
+                                                                         src  + offset,
                                                                          thisRoundNumRows,
                                                                          fRealColumns[i].col.size,
@@ -716,6 +756,6 @@
                         break;
                         case zfits::kFactSmoothing:
-                                applySMOOTHING(&(fCompressedBuffer[threadIndex][compressedOffset]),
-                                               &(fTransposedBuffer[threadIndex][offset]),
+                                applySMOOTHING(dest + compressedOffset,
+                                               src  + offset,
                                                thisRoundNumRows,
                                                fRealColumns[i].col.size,
@@ -724,12 +764,12 @@
                         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,36 +786,37 @@
                     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,6 +826,6 @@
                         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,11 +835,10 @@
                             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,13 +884,4 @@
         }
 
-        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,13 +904,12 @@
         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,5 +920,4 @@
         } __attribute__((__packed__));
 
-       // typedef pair<int64_t, int64_t> CatalogEntry;
         typedef vector<CatalogEntry>   CatalogRow;
         typedef vector<CatalogRow>     CatalogType;
@@ -897,12 +926,14 @@
         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,12 +948,4 @@
         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
-
 };