x3c/include/packetReader/CPacketReader.h

/*
 * CPacketReader.h
 *
 *  Created on: Nov 18, 2011
 *      Author: mtaveniku
 *  The Packet reader class is a multithreaded class reading X3C packets from
 *  a parallel file system. The Packets in the stream are delivered in
 *  chronological sequence from the files
 *  This class is supposed to work on both Windows and Linux machines
 */

#ifndef CPACKETREADER_H_
#define CPACKETREADER_H_

#include <vector>

#include "X3cPacket.h"
#include "CFileReader.h"
#include "CFrameFilter.h"
#include "CTimeFilter.h"

using namespace std;
//#define INTERNAL_PACKET_HANDLING


class CPacketReader {

public:
        static const size_t MAX_DISK_READ_SIZE_MB = 128;

        typedef struct _packet_handle_ {
                int index;
                int streamNo;
                X3cPacket *pkt;
        } packet_handle_t;

public:


        CPacketReader();


        virtual ~CPacketReader();


        X3cPacket *getPacket();


        void freePacket(X3cPacket *pkt);

    void setPacketFilter(UINT64 start, UINT64 end);

    void setTimeFilter(CTimeFilter const& tFilter);


    UINT64 getPacketCount() const;

        // int seekTime(struct timespec ts);
        // int seek(size_t position);
        // int seekPacket(unsigned long pktNo);



        bool initialize(std::vector<std::string> const& fileNames, size_t bufSz,
                    size_t chunkSz);



        int start();


        bool stop();

#ifdef INTERNAL_PACKET_HANDLING


        packet_handle_t *getPacketHandle();
#else


        void getPacketHandle(packet_handle_t *hdl);
#endif


        void freePacketHandle(packet_handle_t *hdl);


    void reset();

private:

        X3cPacket *persistent_get_packet(int streamNo);


        void sortPackets();

        void freePktHandleIdx(int index);

        int  getFreePktHandleIdx();

        void initPktHandles();

    void resetFileReaders();

    void resetPacketBuffers();

    void resumeFileReaders();


        CStreamFilter::FilterReturn getShouldBeTimeFiltered(X3cPacket *pkt);


        bool mIsStarted;


        bool mIsInitialized;


        int mNumInput;


        vector<std::string> mFileNames;


        size_t mRdChunkSz;


        size_t mBufSz;

        vector <CPacketBuffer*> mPacketBuffers;


        vector <CFileReader*> mFileReaders;


        vector <X3cPacket *> mPktPtrs;


        int mOldestStream;


        UINT64 mOldestTs;


        UINT64 mNextTs;


        static const int mFHListSz = 512;

        //A filter on the packet count;
        CFrameFilter mPacketFilter;

    CTimeFilter mTimeFilter;

        UINT64 mPacketCounter;

#ifdef INTERNAL_PACKET_HANDLING

        packet_handle_t *mPacketHandles;

        int *mFreeHandles;

        int mFHHd;

        int mFHTl;
#endif
};

inline void CPacketReader::setPacketFilter(UINT64 start, UINT64 end)
{
    mPacketFilter.setFilter(start, end);
}

inline void CPacketReader::setTimeFilter(CTimeFilter const& tFilter)
{
    mTimeFilter = tFilter;
}

inline UINT64 CPacketReader::getPacketCount() const
{
    return mPacketCounter;
}

#endif /* CPACKETREADER_H_ */