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

---

IOBufferChannel.h

Go to the documentation of this file.

#ifndef Impala_Util_IOBufferChannel_h
#define Impala_Util_IOBufferChannel_h

#include "Util/IOBuffer.h"
#include "Basis/CmdOptions.h"
#include "Util/ChannelPool.h"
#include "Basis/NativeTypeFormats.h"

namespace Impala
{
namespace Util
{


class IOBufferChannel : public IOBuffer
{
public:

    IOBufferChannel(CString fileName, bool readMode, Channel* channel = 0)
    {
        mReadMode = readMode;
        mId = -1;
        if (channel == 0)
        {
            CmdOptions& options = CmdOptions::GetInstance();
            String dataServer = options.GetString("dataServer");
            String passwordFile = options.GetString("passwordFile");
            channel = ChannelPool::Instance().Get(dataServer, passwordFile);
        }
        mMyChannel = channel;
        char* buf = mMyChannel->Buffer();
        sprintf(buf, "openfilebuffer \"%s\" %d\0", fileName.c_str(), readMode);
        mMyChannel->SendRequest(strlen(buf)+1);
        if (mMyChannel->LastSendHadError())
        {
            ILOG_ERROR("Failed to open " << fileName);
            return;
        }
        Int64 bufSize;
        sscanf(buf, "%d %lld", &mId, &bufSize);
        if (mReadMode)
        {
            SetSize(bufSize);
            ILOG_DEBUG(fileName << ", id = " << mId << ", size = " << bufSize);
        }
        else
        {
            ILOG_DEBUG(fileName << ", id = " << mId);
        }
    }

    virtual
    ~IOBufferChannel()
    {
        if (mId != -1)
        {
            char* buf = mMyChannel->Buffer();
            sprintf(buf, "closefilebuffer %d\0", mId);
            mMyChannel->SendRequest(strlen(buf)+1);
            mId = -1;
        }
    }

    virtual bool
    Valid()
    {
        return (mId != -1);
    }

    virtual Int64
    Read(void* buf, Int64 bytesToRead)
    {
        char* chBuf = mMyChannel->Buffer();
        Int64 nrRead = 0;
        char* bufPtr = (char*) buf;
        while (nrRead < bytesToRead)
        {
            int nrData = Min<Int64>(Channel::DATA_BUFFER_SIZE, 
                                    bytesToRead - nrRead);
            sprintf(chBuf, "readfilebuffer %d %d\0", mId, nrData);
            int len = mMyChannel->SendRequest(strlen(chBuf)+1);
            if (mMyChannel->LastSendHadError())
                return 0;
            memcpy(bufPtr + nrRead, chBuf, len);
            if (len == 0)
                break;
            nrRead += len;
        }
        SetPosition(GetPosition() + nrRead);
        return nrRead;
    }

    virtual String
    ReadLine()
    {
        // todo : this is a very inefficient implementation
        Int64 available = Available();
        Int64 maxLineSize = 16384; // 16 Kb, todo : give user control
        if (available < maxLineSize)
            maxLineSize = available + 1; // extra space for '\0'
        char* buf = new char[maxLineSize];
        PositionType start = GetPosition();
        Int64 nrRead = Read(buf, maxLineSize); // does SetPosition
        Int64 i = 0;
        while ((i < maxLineSize) && (buf[i] != '\n'))
            i++;
        Int64 skipN = (buf[i] == '\n') ? 1 : 0;
        SetPosition(start + i + skipN);
        if ((i > 0) && (buf[i-1] == '\r')) // strip '\r' if present
            i--;
        buf[i] = '\0';
        String res(buf);
        delete buf;
        return res;
    }

    virtual Int64
    Gets(char* buf, Int64 bytesToRead)
    {
        Int64 available = Available();
        if (available < bytesToRead)
            bytesToRead = available;
        PositionType start = GetPosition();
        Int64 nrRead = Read(buf, bytesToRead); // does SetPosition
        Int64 i = 0;
        while ((i < bytesToRead) && (buf[i] != '\n'))
            i++;
        bytesToRead = i;
        if (buf[i] == '\n')
            i++;
        SetPosition(start + i);
        return bytesToRead;
    }

    virtual void
    NativeTypeRead(Int8* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual void
    NativeTypeRead(UInt8* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual void
    NativeTypeRead(Int16* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual void
    NativeTypeRead(UInt16* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual void
    NativeTypeRead(Int32* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual void
    NativeTypeRead(UInt32* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual void
    NativeTypeRead(Int64* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual void
    NativeTypeRead(UInt64* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual void
    NativeTypeRead(Real32* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual void
    NativeTypeRead(Real64* ptr)
    {
        DoNativeTypeRead(ptr);
    }

    virtual Int64
    Write(const void* buf, Int64 bytesToWrite)
    {
        char* chBuf = mMyChannel->Buffer();
        Int64 nrWritten = 0;
        char* bufPtr = (char*) buf;
        while (nrWritten < bytesToWrite)
        {
            sprintf(chBuf, "writefilebuffer %d\0", mId);
            int used = strlen(chBuf) + 1;
            int nrData = Min<Int64>(Channel::DATA_BUFFER_SIZE - used,
                                    bytesToWrite - nrWritten);
            memcpy(chBuf + used, bufPtr + nrWritten, nrData);
            int len = mMyChannel->SendRequest(used + nrData);
            if (mMyChannel->LastSendHadError())
                break;
            nrWritten += nrData;
        }
        SetPosition(GetPosition() + bytesToWrite);
        return nrWritten;
    }

    virtual Int64
    Puts(const char* buf)
    {
        Int64 len = strlen(buf) + 1;
        char* writeBuf = new char[len];
        memcpy(writeBuf, buf, len);
        writeBuf[len-1] = '\n';
        Write(writeBuf, len);
        delete writeBuf;
        return 1;
    }

    virtual void
    NativeTypeWrite(Int8 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    NativeTypeWrite(UInt8 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    NativeTypeWrite(Int16 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    NativeTypeWrite(UInt16 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    NativeTypeWrite(Int32 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    NativeTypeWrite(UInt32 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    NativeTypeWrite(Int64 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    NativeTypeWrite(UInt64 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    NativeTypeWrite(Real32 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    NativeTypeWrite(Real64 val)
    {
        DoNativeTypeWrite(val);
    }

    virtual void
    SetPosition(PositionType position)
    {
        if (GetPosition() == position)
            return;

        char* buf = mMyChannel->Buffer();
        sprintf(buf, "seekfilebuffer %d %lld %d\0", mId, position, SEEK_SET);
        mMyChannel->SendRequest(strlen(buf)+1);
        if (mMyChannel->LastSendHadError())
        {
            ILOG_ERROR("Failed to seek");
            return;
        }
        PositionType newPos;
        sscanf(buf, "%lld", &newPos);
        IOBuffer::SetPosition(newPos);
    }

private:

    template <class NativeType>
    void
    DoNativeTypeRead(NativeType* ptr)
    {
        String fs = NativeTypeFormat<NativeType>(0);
        char buf[1024];
        PositionType start = GetPosition();
        Int64 nrRead = Read(buf, 1024); // does SetPosition
        Int64 i = 0;
        while ((buf[i] != ' ') && (i < nrRead))
            i++;
        if(nrRead == 0)
            ILOG_ERROR("DoNativeTypeRead read 0 bytes")
        sscanf(buf, fs.c_str(), ptr);
        SetPosition(start + i + 1);
    }

    template <class NativeType>
    void
    DoNativeTypeWrite(NativeType val)
    {
        String fs = NativeTypeFormat<NativeType>(0) + " ";
        char buf[1024];
        sprintf(buf, fs.c_str(), val);
        Int64 i = 0;
        while (buf[i] != ' ')
            i++;
        Write(buf, i+1);
    }

    bool     mReadMode;
    int      mId;
    Channel* mMyChannel; // watch out for IOBuffer::mDataChannel conflicts

    ILOG_VAR_DEC;
};

ILOG_VAR_INIT(IOBufferChannel, Impala.Util);

} // namespace Util
} // namespace Impala

#endif

---