YiConcurrentDeque.h

Go to the documentation of this file.

// © You i Labs Inc. 2000-2017. All rights reserved.
#ifndef _YI_CONCURRENT_DEQUE_H_
#define _YI_CONCURRENT_DEQUE_H_

#include <deque>

#include "thread/YiAutoMutex.h"
#include "thread/YiMutex.h"
#include "thread/YiWaitCondition.h"

template<typename YI_DATA>
class CYIConcurrentDeque
{
public:
    CYIConcurrentDeque() :
        m_bStopWaiting(false)
    {
    }

    CYIConcurrentDeque(const CYIConcurrentDeque<YI_DATA> &other) :
        m_bStopWaiting(false)
    {
        CYIAutoMutex autoMutex(other.m_mutex);
        m_deque = other.m_deque;
    }

    CYIConcurrentDeque & operator=(const CYIConcurrentDeque<YI_DATA> & rhs)
    {
        CYIAutoMutex autoLhsMutex(m_mutex);
        CYIAutoMutex autoRhsMutex(rhs.m_mutex);
        m_deque = rhs.m_deque;
        return *this;
    }

    void PushFront(const YI_DATA &rData)
    {
        CYIAutoMutex autoMutex(m_mutex);

        m_deque.push_front(rData);

        m_AvailableCondition.WakeOne();
    }

    void PushBack(const YI_DATA &rData)
    {
        CYIAutoMutex autoMutex(m_mutex);

        m_deque.push_back(rData);

        m_AvailableCondition.WakeOne();
    }

    void WaitEmpty()
    {
        CYIAutoMutex autoMutex(m_mutex);
        while (!m_deque.empty()) // protection against 'spurious wake-ups'
        {
            m_EmptyCondition.Wait(m_mutex);
        }
    }

    bool Empty() const
    {
        CYIAutoMutex autoMutex(m_mutex);
        return m_deque.empty();
    }

    bool TryPopFront(YI_DATA & rValue)
    {
        bool bSuccessfulLock = false;

        if (m_mutex.TryLock())
        {
            if(!m_deque.empty())
            {
                rValue = m_deque.front();
                m_deque.pop_front();
                
                bSuccessfulLock = true;
                
                if (m_deque.empty())
                {
                    m_EmptyCondition.WakeAll();
                }
            }

            m_mutex.Unlock();
        }

        return bSuccessfulLock;
    }

    bool TryPopBack(YI_DATA & rValue)
    {
        bool bSuccessfulLock = false;

        if (m_mutex.TryLock())
        {
            if(!m_deque.empty())
            {
                rValue = m_deque.back();
                m_deque.pop_back();
                
                bSuccessfulLock = true;
                
                if (m_deque.empty())
                {
                    m_EmptyCondition.WakeAll();
                }
            }

            m_mutex.Unlock();
        }

        return bSuccessfulLock;
    }

    bool WaitAndPopFront(YI_DATA &rValue)
    {
        bool bRet = false;

        CYIAutoMutex autoMutex(m_mutex);

        while(m_deque.empty() && !m_bStopWaiting)
        {
            m_AvailableCondition.Wait(m_mutex);
        }

        m_bStopWaiting = false;
        
        if (!m_deque.empty())
        {
            rValue = m_deque.front();
            m_deque.pop_front();

            if (m_deque.empty())
            {
                m_EmptyCondition.WakeAll();
            }

            bRet = true;
        }

        return bRet;
    }

    bool WaitAndPopFront(YI_DATA &rValue, uint64_t uTimeoutMs)
    {
        bool bRet = false;

        CYIAutoMutex autoMutex(m_mutex);

        if (!m_bStopWaiting && m_deque.empty())
        {
            m_AvailableCondition.Wait(m_mutex, uTimeoutMs);
        }

        m_bStopWaiting = false;

        if (!m_deque.empty())
        {
            rValue = m_deque.front();
            m_deque.pop_front();

            if (m_deque.empty())
            {
                m_EmptyCondition.WakeAll();
            }

            bRet = true;
        }
        
        return bRet;
    }

    bool WaitAndPopBack(YI_DATA & rValue)
    {
        bool bRet = false;

        CYIAutoMutex autoMutex(m_mutex);

        while(m_deque.empty() && !m_bStopWaiting)
        {
            m_AvailableCondition.Wait(m_mutex);
        }

        m_bStopWaiting = false;
        
        if (!m_deque.empty())
        {
            rValue = m_deque.back();
            m_deque.pop_back();

            if (m_deque.empty())
            {
                m_EmptyCondition.WakeAll();
            }

            bRet = true;
        }

        return bRet;
    }

    void StopWaiting()
    {
        CYIAutoMutex autoMutex(m_mutex);

        m_bStopWaiting = true;
        m_AvailableCondition.WakeOne();
    }

    uint32_t GetSize() const
    {
        CYIAutoMutex autoMutex(m_mutex);
        return (uint32_t)m_deque.size();
    }

private:
    std::deque<YI_DATA> m_deque;
    mutable CYIMutex m_mutex;

    bool m_bStopWaiting;
    CYIWaitCondition m_EmptyCondition;
    CYIWaitCondition m_AvailableCondition;

};

#endif // _YI_CONCURRENT_DEQUE_H_