uniquequeue.h

Go to the documentation of this file.

/*
    Copyright (C) 2020 by Pawel Soja <[email protected]>
 
    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2.1 of the License, or (at your option) any later version.
    This library is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
    Lesser General Public License for more details.
    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
*/
#pragma once
 
#include <queue>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include <cstdint>
 
template <typename T>
class UniqueQueue
{
    public:
        void push(T &&data);
 
        bool pop(T &dest);
 
        bool pop(T &dest, uint32_t msecs);
 
        void waitForEmpty() const;
 
        bool waitForEmpty(uint32_t msecs) const;
 
        void clear();
 
        void abort();
 
        size_t size() const;
 
    protected:
        std::queue<T>      queue;
        mutable std::mutex mutex;
 
        mutable std::condition_variable decrease;
        mutable std::condition_variable increase;
};
 
// implementation
template <typename T>
inline void UniqueQueue<T>::push(T &&data)
{
    std::lock_guard<std::mutex> lock(mutex);
    queue.push(std::move(data));
    increase.notify_all();
}
 
template <typename T>
inline bool UniqueQueue<T>::pop(T &dest)
{
    std::unique_lock<std::mutex> lock(mutex);
    if (queue.empty())
        increase.wait(lock);
 
    if (queue.empty())
        return false; // abort
 
    std::swap(dest, queue.front());
    queue.pop();
    decrease.notify_all();
    return true;
}
 
template <typename T>
inline bool UniqueQueue<T>::pop(T &dest, uint32_t msecs)
{
    std::unique_lock<std::mutex> lock(mutex);
 
    if (queue.empty() && increase.wait_for(lock, std::chrono::milliseconds(msecs)) == std::cv_status::timeout)
        return false; // timeout
 
    if (queue.empty())
        return false; // abort
 
    std::swap(dest, queue.front());
    queue.pop();
    decrease.notify_all();
    return true;
}
 
template <typename T>
inline size_t UniqueQueue<T>::size() const
{
    std::unique_lock<std::mutex> lock(mutex);
    return queue.size();
}
 
template <typename T>
inline void UniqueQueue<T>::clear()
{
    std::queue<T> empty;
    std::unique_lock<std::mutex> lock(mutex);
    std::swap(queue, empty);
    decrease.notify_all();
}
 
template <typename T>
inline void UniqueQueue<T>::waitForEmpty() const
{
    std::unique_lock<std::mutex> lock(mutex);
    decrease.wait(lock, [this]()
    {
        return queue.empty();
    });
}
 
template <typename T>
inline bool UniqueQueue<T>::waitForEmpty(uint32_t msecs) const
{
    std::unique_lock<std::mutex> lock(mutex);
    return decrease.wait_for(lock, std::chrono::milliseconds(msecs), [this]()
    {
        return queue.empty();
    });
}
 
template <typename T>
inline void UniqueQueue<T>::abort()
{
    std::queue<T> empty;
    std::unique_lock<std::mutex> lock(mutex);
    std::swap(queue, empty);
    increase.notify_all();
    decrease.notify_all();
}