YiConcurrentQueue.h

Go to the documentation of this file.

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

#include <queue>

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

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

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

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

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

        m_queue.push(rData);

        m_AvailableCondition.WakeOne();
    }

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

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

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

        if (m_mutex.TryLock())
        {
            if(!m_queue.empty())
            {
                rValue = m_queue.front();
                m_queue.pop();
                
                bSuccessfulLock = true;
                
                if (m_queue.empty())
                {
                    m_EmptyCondition.WakeAll();
                }
            }
            
            m_mutex.Unlock();
        }

        return bSuccessfulLock;
    }

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

        CYIAutoMutex autoMutex(m_mutex);

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

        m_bStopWaiting = false;
        
        if (!m_queue.empty())
        {
            rValue = m_queue.front();
            m_queue.pop();

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

            bRet = true;
        }

        return bRet;
    }

    void StopWaiting()
    {
        CYIAutoMutex autoMutex(m_mutex);

        m_bStopWaiting = true;
        m_AvailableCondition.WakeOne();
    }

private:
    std::queue<YI_DATA> m_queue;
    mutable CYIMutex m_mutex;

    bool m_bStopWaiting;
    CYIWaitCondition m_EmptyCondition;
    CYIWaitCondition m_AvailableCondition;

};

#endif // _YI_CONCURRENT_QUEUE_H_