queueBase.hpp

Go to the documentation of this file.

#ifndef OPENPOSE_THREAD_QUEUE_BASE_HPP
#define OPENPOSE_THREAD_QUEUE_BASE_HPP

#include <condition_variable>
#include <mutex>
#include <queue> // std::queue & std::priority_queue
#include <openpose/core/common.hpp>

namespace op
{
    template<typename TDatums, typename TQueue>
    class QueueBase
    {
    public:
        explicit QueueBase(const long long maxSize = -1);

        virtual ~QueueBase();

        bool forceEmplace(TDatums& tDatums);

        bool tryEmplace(TDatums& tDatums);

        bool waitAndEmplace(TDatums& tDatums);

        bool forcePush(const TDatums& tDatums);

        bool tryPush(const TDatums& tDatums);

        bool waitAndPush(const TDatums& tDatums);

        bool tryPop(TDatums& tDatums);

        bool tryPop();

        bool waitAndPop(TDatums& tDatums);

        bool waitAndPop();

        bool empty() const;

        void stop();

        void stopPusher();

        void addPopper();

        void addPusher();

        bool isRunning() const;

        size_t size() const;

        void clear();

        virtual TDatums front() const = 0;

    protected:
        mutable std::mutex mMutex;
        long long mPoppers;
        long long mPushers;
        long long mMaxPoppersPushers;
        bool mPopIsStopped;
        bool mPushIsStopped;
        std::condition_variable mConditionVariable;
        TQueue mTQueue;

        virtual bool pop(TDatums& tDatums) = 0;

        unsigned long long getMaxSize() const;

    private:
        const long long mMaxSize;

        bool emplace(TDatums& tDatums);

        bool push(const TDatums& tDatums);

        bool pop();

        void updateMaxPoppersPushers();

        DELETE_COPY(QueueBase);
    };
}





// Implementation
#include <openpose/core/datum.hpp>
#include <openpose/utilities/fastMath.hpp>
namespace op
{
    template<typename TDatums, typename TQueue>
    QueueBase<TDatums, TQueue>::QueueBase(const long long maxSize) :
        mPoppers{0ll},
        mPushers{0ll},
        mPopIsStopped{false},
        mPushIsStopped{false},
        mMaxSize{maxSize}
    {
    }

    // Virutal destructor
    template<typename TDatums, typename TQueue>
    QueueBase<TDatums, TQueue>::~QueueBase()
    {
        try
        {
            log("", Priority::Low, __LINE__, __FUNCTION__, __FILE__);
            stop();
            log("", Priority::Low, __LINE__, __FUNCTION__, __FILE__);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::forceEmplace(TDatums& tDatums)
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            if (mTQueue.size() >= getMaxSize())
                mTQueue.pop();
            return emplace(tDatums);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::tryEmplace(TDatums& tDatums)
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            if (mTQueue.size() >= getMaxSize())
                return false;
            return emplace(tDatums);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::waitAndEmplace(TDatums& tDatums)
    {
        try
        {
            std::unique_lock<std::mutex> lock{mMutex};
            mConditionVariable.wait(lock, [this]{return mTQueue.size() < getMaxSize() || mPushIsStopped; });
            return emplace(tDatums);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::forcePush(const TDatums& tDatums)
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            if (mTQueue.size() >= getMaxSize())
                mTQueue.pop();
            return push(tDatums);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::tryPush(const TDatums& tDatums)
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            if (mTQueue.size() >= getMaxSize())
                return false;
            return push(tDatums);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::waitAndPush(const TDatums& tDatums)
    {
        try
        {
            std::unique_lock<std::mutex> lock{mMutex};
            mConditionVariable.wait(lock, [this]{return mTQueue.size() < getMaxSize() || mPushIsStopped; });
            return push(tDatums);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::tryPop(TDatums& tDatums)
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            return pop(tDatums);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::tryPop()
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            return pop();
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::waitAndPop(TDatums& tDatums)
    {
        try
        {
            std::unique_lock<std::mutex> lock{mMutex};
            mConditionVariable.wait(lock, [this]{return !mTQueue.empty() || mPopIsStopped; });
            return pop(tDatums);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::waitAndPop()
    {
        try
        {
            std::unique_lock<std::mutex> lock{mMutex};
            mConditionVariable.wait(lock, [this]{return !mTQueue.empty() || mPopIsStopped; });
            return pop();
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::empty() const
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            return mTQueue.empty();
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    void QueueBase<TDatums, TQueue>::stop()
    {
        try
        {
            log("", Priority::Low, __LINE__, __FUNCTION__, __FILE__);
            const std::lock_guard<std::mutex> lock{mMutex};
            mPopIsStopped = {true};
            mPushIsStopped = {true};
            while (!mTQueue.empty())
                mTQueue.pop();
            mConditionVariable.notify_all();
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
        }
    }

    template<typename TDatums, typename TQueue>
    void QueueBase<TDatums, TQueue>::stopPusher()
    {
        try
        {
            log("", Priority::Low, __LINE__, __FUNCTION__, __FILE__);
            const std::lock_guard<std::mutex> lock{mMutex};
            mPushers--;
            if (mPushers == 0)
            {
                mPushIsStopped = {true};
                if (mTQueue.empty())
                    mPopIsStopped = {true};
                mConditionVariable.notify_all();
            }
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
        }
    }

    template<typename TDatums, typename TQueue>
    void QueueBase<TDatums, TQueue>::addPopper()
    {
        try
        {
            log("", Priority::Low, __LINE__, __FUNCTION__, __FILE__);
            const std::lock_guard<std::mutex> lock{mMutex};
            mPoppers++;
            updateMaxPoppersPushers();
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
        }
    }

    template<typename TDatums, typename TQueue>
    void QueueBase<TDatums, TQueue>::addPusher()
    {
        try
        {
            log("", Priority::Low, __LINE__, __FUNCTION__, __FILE__);
            const std::lock_guard<std::mutex> lock{mMutex};
            mPushers++;
            updateMaxPoppersPushers();
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::isRunning() const
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            return !(mPushIsStopped && (mPopIsStopped || mTQueue.empty()));
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return true;
        }
    }

    template<typename TDatums, typename TQueue>
    size_t QueueBase<TDatums, TQueue>::size() const
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            return mTQueue.size();
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return 0;
        }
    }

    template<typename TDatums, typename TQueue>
    void QueueBase<TDatums, TQueue>::clear()
    {
        try
        {
            const std::lock_guard<std::mutex> lock{mMutex};
            while (!mTQueue.empty())
                mTQueue.pop();
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
        }
    }

    template<typename TDatums, typename TQueue>
    unsigned long long QueueBase<TDatums, TQueue>::getMaxSize() const
    {
        try
        {
            return (mMaxSize > 0 ? mMaxSize : fastMax(1ll, mMaxPoppersPushers));
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::emplace(TDatums& tDatums)
    {
        try
        {
            if (mPushIsStopped)
                return false;

            mTQueue.emplace(tDatums);
            mConditionVariable.notify_all();
            return true;
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::push(const TDatums& tDatums)
    {
        try
        {
            if (mPushIsStopped)
                return false;

            mTQueue.push(tDatums);
            mConditionVariable.notify_all();
            return true;
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    bool QueueBase<TDatums, TQueue>::pop()
    {
        try
        {
            if (mPopIsStopped || mTQueue.empty())
                return false;

            mTQueue.pop();
            mConditionVariable.notify_all();
            return true;
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
            return false;
        }
    }

    template<typename TDatums, typename TQueue>
    void QueueBase<TDatums, TQueue>::updateMaxPoppersPushers()
    {
        try
        {
            mMaxPoppersPushers = fastMax(mPoppers, mPushers);
        }
        catch (const std::exception& e)
        {
            error(e.what(), __LINE__, __FUNCTION__, __FILE__);
        }
    }

    extern template class QueueBase<DATUM_BASE, std::queue<DATUM_BASE>>;
    extern template class QueueBase<DATUM_BASE, std::priority_queue<DATUM_BASE, std::vector<DATUM_BASE>, std::greater<DATUM_BASE>>>;
}

#endif // OPENPOSE_THREAD_QUEUE_BASE_HPP