IPQbufferBase.h

Go to the documentation of this file.

#ifndef CARMA_UTIL_IPQBUFFERBASE_H
#define CARMA_UTIL_IPQBUFFERBASE_H

#include "carma/util/IPQinterface.h"

#include <string>

namespace carma {
    namespace util {

    // Forward declaration
    class PthreadRWLock;
    class ScopedFlockManager;

class IPQbufferBase : public carma::util::IPQinterface {
protected:
    IPQbufferBase( void *                localElement,
                   int                   elementSize,
                   const ::std::string&  filename,
                   bool                  isCreator = false,
                   int                   nElements = 0,
                   unsigned int          testOffset = 0 );
public:
    virtual ~IPQbufferBase() ;

    ::std::string getFileName( ) const;

    int getQueueSize() const ;

    int getElementSize() const ;

    int getNumAvailable() const ;

    bool isEmpty() const ;

    bool isDataAvailable() const;

    void setNoneAvailable();

    unsigned int read();


private:

    unsigned int readNoLock();

public:

    bool readNewest();
    
    bool readNewestConditionalCopy();
    
private:

    bool readNewestNoLock();

public:

    unsigned int getLostElementCount() const;

    unsigned int getPutOffset() const ;
    unsigned int getGetOffset() const ;
    unsigned int getMaxOffset() const ;

protected:

    void init() ;

    void write() ;

    virtual bool openBuffer() = 0;

    virtual bool createBuffer() = 0;

    void trimShmemFilename();

    ::std::string getTrimmedFilename();

private:
    void copyMemory( void * const       dest,
                     const void * const src,
                     const size_t       bytes ) const;

    bool readHeader(bool keepOpen) ;

    bool doesElementSizeMatch() ;

    bool hasQueueChanged() ;

    bool writeHeader() ;

    void createSemFilename();

    bool openSemaphore();

    bool createSemaphore();
    
    // Internal routine that handles wraparound.
    // Will lock the flock manager if the read/write pointers are adjacent
    // returns adjusted readoffset
    unsigned int adjustReadOffset(bool& lock, ScopedFlockManager& flockManager, 
        unsigned int readOffset);

    unsigned int   internalRead(bool lock) ;
    bool           internalReadNewest(bool lock) ;

private:
    class BlockedOnSemaphoreQuitRequestHandler;

    friend class BlockedOnSemaphoreQuitRequestHandler;

    static const int MEMORYMAPPEDFILE_MAGIC = 0x00000100;

    struct ControlBlock {
        // An index for the next element to be written.
        // Steadily increments through the full range of a long,
        // with modulos used to put it into the finite queue
        volatile unsigned int putOffset;
    };

    struct MapfileHeader {
        int          magic;
        unsigned int headerSize;
        unsigned int controlBlockOffset;
        unsigned int controlBlockSize;
        // Offset in bytes from the start of the shmem to start of queue area
        unsigned int queueOffset;
        // The size of a single element in the queue, in bytes 
        unsigned int queueElementSize; 
        // The number of elements allocated for the queue
        unsigned int queueElements;
    };

protected:
    const int protectionMask_;
    const int openMask_;

private:
    const ::std::string filename_;         // As input

    // The local copy of the data in an element.
    // It has been read from the queue and placed here - or it is
    // the copy that will be transferred into the queue on writes.
    void * const        localElement_;
    const size_t        localElementSize_;
    const bool          isCreator_;

protected:
    bool                debug_;
    int                 fileDescriptor_;

private:
    int                 semDescriptor_;  // Semaphore descriptor...

    void*               mmapAddr_;
    size_t              mmapSize_;

    size_t                 controlBlockOffset_;
    volatile ControlBlock* controlBlock_;

    // Offset in bytes from the start of the shmem area to start of queue area
    size_t              queueOffset_;
    // Size of an element in bytes
    size_t              queueElementSize_;
    // Number of elements allocated for the queue
    unsigned int        queueElements_;
    char*               queue_; // Absolute pointe to beginning of the queue

    bool                isNewFile_;
    bool                isFileOpen_;

    // Offset where next data will be read
    unsigned int        getOffset_;
    unsigned int        nLostElements_;

    ::std::string       trimmedFilename_;
    ::std::string       displayFilename_;
    ::std::string       semFilename_;      // Semaphore filename

    MapfileHeader       header_;
    PthreadRWLock&      rwLock_;
    unsigned int        maxOffset_;  // Max for put/get pointers (wraparound)
    // To help resolve emptiness at wraparound time
    bool                empty_;
    unsigned int        testOffset_;
};

} }  // End namespace carma::util

#endif  // CARMA_UTIL_IPQBUFFERBASE_H