TnbQueue.h

[詳解]

#pragma once
#include "TnbVector.h"
 
 
 
//TNB Library
namespace TNB
{
 
 
 
template<typename TYP>
struct IQueueT
{
    virtual ~IQueueT(void) 
    {
    }
 
    virtual size_t AddElements(size_t size, const TYP* P = NULL) = 0;
 
    virtual INDEX Add(const TYP& t) = 0;
 
    virtual size_t TakeElements(size_t size, TYP* P = NULL) = 0;
 
    virtual TYP Take(void)
    {
        TYP t;
        if ( TakeElements(1, &t) == 1 )
        {
            return t;
        }
        ASSERT0( false, "IQueue::Take()", "キューが空の時取得操作されました。" );
        throw CEmptyException();
    }
 
    virtual size_t EraseElements(size_t size)
    {
        return TakeElements(size);
    }
 
    CVectorT<TYP> TakeoutToVector(size_t size)
    {
        CVectorT<TYP> v;
        bool boRc = (TakeElements(size, v.GetBuffer(size)) == size);
        v.ReleaseBuffer();
        if ( ! boRc )
        {
            v.Invalid();
        }
        return v;
    }
};
 
 
 
template<typename TYP, bool ISI = false, class INV = TYP>
class CRingQueueT : public ICollectionMidT<TYP>, public IQueueT<TYP>
{
    DEFSUPER(ICollectionMidT<TYP>);
public:
 
    explicit CRingQueueT(size_t size = 10000) : _super(), m_hptHead(new THead(size))
    {
    }
 
    CRingQueueT(const CRingQueueT& queue) : _super(), m_hptHead(queue.m_hptHead)
    {
    }
    
    virtual ~CRingQueueT(void)
    {
        m_hptHead.Null();
    }
 
    CRingQueueT& operator=(const CRingQueueT& other)
    {
        EXCLUSIVE(&other);
        m_hptHead = other.m_hptHead;
        return *this;
    }
 
    virtual bool Lock(DWORD dwTime = INFINITE) const
    {
        return m_syncFunc.Lock();
    }
 
    virtual void Unlock(void) const
    {
        m_syncFunc.Unlock();
    }
 
    void SetQueueSize(size_t size)
    {
        m_hptHead = new THead(size);
        ASSERT( GetSize() == 0 );
    }
 
    size_t GetQueueSize(void) const
    {
        return m_hptHead->bufferSize;
    }
 
    virtual size_t GetSize(void) const 
    {
        return m_hptHead->dataSize;
    }
 
    bool IsFull(void) const
    {
        if ( GetQueueSize() <= GetSize() )
        {
            return true;
        }
        return false;
    }
 
    virtual INDEX Add(const TYP& t)
    {
        size_t r = GetSize();
        return (AddElements(1, &t) == 1) ? r : INVALID_INDEX;
    }
 
    virtual bool Remove(INDEX index)
    {
        if ( IsInRange(index) )
        {
            m_Separate();
            THead* H = *m_hptHead.ReferP();
            size_t sz = H->dataSize;
            INDEX ri = H->readIndex;
            size_t bs   = H->bufferSize;
            loop ( i, sz -index - 1 )
            {
                INDEX dst = (ri + i + index) % bs;
                INDEX src = (ri + i + index + 1) % bs;
                H->pBuffer[dst] = H->pBuffer[src];
            }
            if ( ISI )
            {
                INDEX dst = (ri + sz - 1) % bs;
                H->pBuffer[dst] = INV();
            }
            (H->dataSize)--;
            return true;
        }
        return false; 
    }
 
    virtual bool RemoveAll(void)
    {
        EXCLUSIVE(this);
        size_t l = m_hptHead->dataSize;
        if ( l > 0 )
        {
            TakeElements(l);
            ASSERT( m_hptHead->dataSize == 0 );
        }
        return true;
    }
 
    virtual size_t AddElements(size_t size, const TYP* P = NULL)
    {
        EXCLUSIVE(this);
        m_Separate();
        THead* H = *m_hptHead.ReferP();
        if ( (H->dataSize) + size > (H->bufferSize) )
        {
            //それ以上になるならエラー
            return INVALID_SIZE;
        }
        H->syncObj.Lock();
        size_t bs   = H->bufferSize;
        INDEX tmp = (H->readIndex + H->dataSize) % bs;
        if ( P != NULL )
        {
            loop ( i, size )
            {
                H->pBuffer[tmp++] = *P++;
                if ( tmp >= bs )
                {
                    tmp -= bs;
                }
            }
        }
        else
        {
            loop ( i, size )
            {
                H->pBuffer[tmp++] = TYP();
                if ( tmp >= bs )
                {
                    tmp -= bs;
                }
            }
        }
        (H->dataSize) += size;
        H->syncObj.Unlock();
        return size;
    }
 
    virtual size_t TakeElements(size_t size, TYP* P = NULL)
    {
        EXCLUSIVE(this);
        m_Separate();
        THead* H = *m_hptHead.ReferP();
        H->syncObj.Lock();
        if ( (H->dataSize) < size )
        {
            //キューにその長さの情報がないのでエラー
            H->syncObj.Unlock();
            return INVALID_SIZE;
        }
        size_t bs = H->bufferSize;
        if ( P != NULL || ISI )
        {
            INDEX tmp = H->readIndex;
            loop ( i, size )
            {
                if ( P != NULL )
                {
                    *P++ = H->pBuffer[tmp];
                }
                if ( ISI )
                {
                    H->pBuffer[tmp] = INV();
                }
                if ( ++tmp >= bs )
                {
                    tmp -= bs;
                }
            }
        }
        (H->readIndex) += size;
        (H->readIndex) %= bs;
        (H->dataSize) -= size;
        H->syncObj.Unlock();
        return size;
    }
 
    size_t GetFreeSize(void) const
    {
        return (m_hptHead->bufferSize) - (m_hptHead->dataSize);
    }
 
protected:
 
    struct THead
    {
        CSyncSection    syncObj;        
        CSyncEvent      syncTaking;     
        CSyncEvent      syncAdding;     
        size_t          bufferSize;     
        size_t          dataSize;       
        INDEX           readIndex;      
        TYP*            pBuffer;        
        THead(size_t size) : readIndex(0), bufferSize(size), dataSize(0)
        {
            pBuffer = new TYP[size];
        }
        ~THead(void)
        {
            if ( pBuffer != NULL )
            {
                delete[] pBuffer; 
            }
        }
    };
 
    CPointerHandleT<THead>  m_hptHead;          
 
private:
 
    CSyncSection            m_syncFunc;         
 
    void m_Separate(void)
    {
        if ( m_hptHead.GetReferCount() > 1 )
        {
            size_t size = m_hptHead->bufferSize;
            THead* pNew = new THead(size);
            if ( (m_hptHead->dataSize) > 0 )
            {
                pNew->dataSize  = m_hptHead->dataSize;
                pNew->readIndex = m_hptHead->readIndex;
                TYP* P1 = pNew->pBuffer;
                TYP* P2 = m_hptHead->pBuffer;
                while ( (size--) > 0 )
                {
                    *P1++ = *P2++;
                }
            }
            m_hptHead = pNew;
        }
    }
 
    virtual const TYP* m_GetConstPointer(INDEX index) const
    {
        if ( IsInRange(index) )
        {
            INDEX tmp;
            THead* H = *m_hptHead.ReferP();
            tmp = (H->readIndex) + index;
            tmp %= (H->bufferSize);
            return &(H->pBuffer[tmp]);
        }
        return NULL;
    }
 
    virtual TYP* m_GetPointer(INDEX index)
    {
        if ( IsInRange(index) )
        {
            m_Separate();
            INDEX tmp;
            THead* H = *m_hptHead.ReferP();
            tmp = (H->readIndex) + index;
            tmp %= (H->bufferSize);
            return &(H->pBuffer[tmp]);
        }
        return NULL;
    }
 
    friend class CQueueTest;
};
 
 
 
template<typename TYP, bool ISI = false, class INV = TYP>
class CBlockingQueueT : public CRingQueueT<TYP, ISI, INV>
{
    typedef CRingQueueT<TYP, ISI, INV> _super;  
    DEFSUPER(_super);
public:
 
    explicit CBlockingQueueT(size_t size = 10000) : _super(size)
    {
    }
 
    virtual ~CBlockingQueueT(void)
    {
    }
 
    virtual size_t TakeElements(size_t size, TYP* P = NULL)
    {
        if ( _super::TakeElements(size, P) != size )
        {
            //なかった
            while ( true )
            {
                m_hptHead->syncAdding.Lock();
                if ( _super::TakeElements(size, P) == size )
                {
                    break;
                }
            }
        }
        m_hptHead->syncTaking.Pulse();
        return size;
    }
 
    virtual size_t AddElements(size_t size, const TYP* P = NULL)
    {
        if ( _super::AddElements(size, P) != size )
        {
            //追加できなかった
            loop ( i, size )
            {
                while ( true )
                {
                    if ( _super::AddElements(1, P) == 1 )
                    {
                        break;
                    }
                    ::Sleep(10);
                }
                m_hptHead->syncAdding.Pulse();
                P++;
            }
        }
        m_hptHead->syncAdding.Pulse();
        return size;
    }
};
 
 
 
typedef TNB::CRingQueueT<BYTE> CByteQueue;
 
 
 
typedef TNB::CBlockingQueueT<BYTE> CByteBlockingQueue;
 
 
 
typedef TNB::CRingQueueT<DWORD> CDwordQueue;
 
 
 
typedef TNB::CRingQueueT<CStr,true> CStrQueue;
 
 
 
typedef TNB::CBlockingQueueT<CStr,true> CStrBlockingQueue;
 
 
 
#ifdef __AFX_H__
 
    typedef TNB::CRingQueueT<CString,true> CStringQueue;
 
#endif
 
 
 
}; // TNB