|| || || || || ||

---

VirtualMatrixIOBufferReader.h

Go to the documentation of this file.

#ifndef Impala_Core_Matrix_VirtualMatrixIOBufferReader_h
#define Impala_Core_Matrix_VirtualMatrixIOBufferReader_h

#include "Core/Matrix/VirtualMatrix.h"
#include "Util/IOBuffer.h"
#include "Core/Array/ReadRaw.h"

namespace Impala
{
namespace Core
{
namespace Matrix
{


template<class ArrayT>
class VirtualMatrixIOBufferReader : public VirtualMatrix
{
public:

    typedef typename ArrayT::StorType StorT;

    VirtualMatrixIOBufferReader(Util::IOBuffer* buffer, bool ownBuffer)
    {
        Init(buffer, ownBuffer);
    }

    virtual
    ~VirtualMatrixIOBufferReader()
    {
        delete mChunkBuffer;
        mChunkBuffer = 0;
        delete mChunkBufferWrapper;
        mChunkBufferWrapper = 0;
        delete mChunkOffsets;
        mChunkOffsets = 0;
        if (mOwnBuffer)
        {
            delete mIOBuffer;
            mIOBuffer = 0;
        }
    }

    virtual int
    NrRow()
    {
        return mHeight;
    }

    virtual int
    NrCol()
    {
        return mWidth;
    }

private:

    virtual bool
    HasGetRowImplReal64()
    {
        // Actually, we can do both but we have a preference
        return (sizeof(StorT) == 8);
    }

    virtual int
    GetRowImpl(int rowNr, Real64* buffer, int bufferSize)
    {
        return ReadRow(rowNr, buffer, bufferSize);
    }

    virtual bool
    HasGetRowImplReal32()
    {
        // Actually, we can do both but we have a preference
        return (sizeof(StorT) == 4);
    }

    virtual int
    GetRowImpl(int rowNr, Real32* buffer, int bufferSize)
    {
        return ReadRow(rowNr, buffer, bufferSize);
    }

    virtual int
    GetDiagonalImpl(Real64* buffer, int bufferSize)
    {
        for (int i=0 ; i<NrRow() ; i++)
        {
            StorT* chunkBufferLine = SwitchToRow(i);
            buffer[i] = static_cast<Real64>(chunkBufferLine[i]);
        }
        return NrRow();
    }

    template<class OutputT>
    int
    ReadRow(int rowNr, OutputT* buffer, int bufferSize)
    {
        StorT* chunkBufferLine = SwitchToRow(rowNr);
        for (int i=0 ; i<mWidth ; i++)
        {
            buffer[i] = static_cast<OutputT>(chunkBufferLine[i]);
        }
        return mWidth;
    }

    StorT*
    SwitchToRow(int rowNr)
    {
        int chunkNo = rowNr / mLinesPerChunk;
        SwitchToChunk(chunkNo);
        int chunkBufferLineNr = rowNr - mCurrentChunk * mLinesPerChunk;
        return mChunkBuffer + (chunkBufferLineNr * mWidth);
    }

    void
    SwitchToChunk(Int64 chunkNo)
    {
        if (mCurrentChunk == chunkNo)
            return;
        if (chunkNo >= mChunkCount)
        {
            ILOG_ERROR("Invalid chunkNo " << chunkNo << " " << mChunkCount);
            return;
        }
        Int64 nrBytes = mChunkSize;
        if (chunkNo == mChunkCount - 1)
        {
            // last chunk might be smaller
            Int64 linesLeft = NrRow() - mLinesPerChunk * chunkNo;
            nrBytes = linesLeft * sizeof(StorT) * mWidth;
        }
        if (mLinesPerCompressedBlock)
        {
            if (mChunkOffsets[chunkNo] == -1)
            {
                for (Int64 i=1 ; i<chunkNo ; i++)
                {
                    if (mChunkOffsets[i] == -1)
                    {
                        SwitchToChunk(i-1);
                    }
                }
            }
            mIOBuffer->SetPosition(mChunkOffsets[chunkNo]);
            Int64 nrRead = 0;
            if (mCompressor == "zlib")
                nrRead = mIOBuffer->ReadZlib(mChunkBuffer, nrBytes);
            if (mCompressor == "bzip")
                nrRead = mIOBuffer->ReadBzip(mChunkBuffer, nrBytes);
            if (nrRead != nrBytes)
            {
                ILOG_ERROR("readCompressed " << nrRead << " bytes instead of "
                            << nrBytes);
            }
            mChunkOffsets[chunkNo + 1] = mIOBuffer->GetPosition();
        }
        else
        {
            mIOBuffer->SetPosition(mChunkOffsets[chunkNo]);
            Int64 nrRead = mIOBuffer->Read(mChunkBuffer, nrBytes);
            if (nrRead != nrBytes)
            {
                ILOG_ERROR("read " << nrRead << " bytes instead of " << nrBytes);
            }
        }
        Array::Endian(mChunkBufferWrapper, mChunkBufferWrapper);
        mCurrentChunk = chunkNo;
    }
    
    void
    Init(Util::IOBuffer* buffer, bool ownBuffer)
    {
        mIOBuffer = buffer;
        mOwnBuffer = ownBuffer;
        mBaseOffset = buffer->GetPosition();
        mChunkOffsets = 0;
        mChunkBuffer = 0;
        if (! Array::ReadRawHeader<ArrayT>
                (mIOBuffer, &mVersion, &mBinary, &mElemSize, &mWidth, &mHeight,
                 &mNrA, mCompressor, &mLinesPerCompressedBlock))
        {
            ILOG_ERROR("Could not parse header");
            return;
        }
        if (!mBinary)
        {
            ILOG_ERROR("Only binary files are supported");
        }
        if (mElemSize != 1)
        {
            ILOG_ERROR("IOBuffer does not contain a Matrix");
        }

        Int64 lineSize = sizeof(StorT) * mWidth;
        buffer->SetPosition(mBaseOffset + 200);
        if (mLinesPerCompressedBlock > 0)
        {
            mLinesPerChunk = mLinesPerCompressedBlock;
        }
        else
        {
            // uncompressed file: still use the same chunk size
            mLinesPerChunk = Util::IOBuffer::cBzipBlockSize / lineSize;
        }
        mChunkCount = mHeight / mLinesPerChunk;
        if (mHeight % mLinesPerChunk != 0)
            mChunkCount++;
        mChunkSize = lineSize * mLinesPerChunk;
        mChunkOffsets = new Int64[mChunkCount+1];
        mChunkOffsets[0] = mBaseOffset + 200;
        for (Int64 i=1 ; i<mChunkCount+1 ; i++)
        {
            if (mLinesPerCompressedBlock == 0)
            {
                mChunkOffsets[i] = mChunkOffsets[i-1] + mChunkSize;
            }
            else
            {
                mChunkOffsets[i] = -1;
            }
        }

        mChunkBuffer = new StorT[mWidth * mLinesPerChunk];
        mChunkBufferWrapper = new ArrayT(mWidth * mLinesPerChunk, 1, 0, 0,
                                         mChunkBuffer, true);
        mCurrentChunk = -1;
        SwitchToChunk(0);
    }

    Util::IOBuffer* mIOBuffer;
    bool   mOwnBuffer;
    Int64  mBaseOffset;
    Int64* mChunkOffsets;
    Int64  mChunkSize;  // of the uncompressed data within one chunk
    Int64  mChunkCount;
    Int64  mLinesPerChunk;

    // current cache
    Int64   mCurrentChunk;
    StorT*  mChunkBuffer;
    ArrayT* mChunkBufferWrapper;

    Int64  mVersion;
    Int64  mBinary;
    Int64  mElemSize;
    Int64  mWidth;
    Int64  mHeight;
    Int64  mNrA;
    String mCompressor;
    Int64  mLinesPerCompressedBlock;

    ILOG_VAR_DEC;
};

ILOG_VAR_INIT_TEMPL_1(VirtualMatrixIOBufferReader, ArrayT, Impala.Core.Matrix);

} // namespace Matrix
} // namespace Core
} // namespace Impala

#endif

---