queue.h

Go to the documentation of this file.

 
#pragma once
 
 
#include <promeki/config.h>
#if PROMEKI_ENABLE_CORE
#include <queue>
#include <promeki/namespace.h>
#include <promeki/error.h>
#include <promeki/list.h>
#include <promeki/mutex.h>
#include <promeki/result.h>
#include <promeki/waitcondition.h>
 
PROMEKI_NAMESPACE_BEGIN
 
template <typename T> class Queue {
        public:
                Queue() = default;
                ~Queue() = default;
 
                void setMaxSize(size_t n) {
                        Mutex::Locker locker(_mutex);
                        _maxSize = n;
                        // A larger / disabled cap may have just made room for
                        // producers waiting on a smaller cap that was set
                        // earlier.  No-op when no producer is parked.
                        _notFull.wakeAll();
                }
 
                size_t maxSize() const {
                        Mutex::Locker locker(_mutex);
                        return _maxSize;
                }
 
                void push(const T &val) {
                        Mutex::Locker locker(_mutex);
                        _queue.push(val);
                        _cv.wakeOne();
                        return;
                }
 
                void push(T &&val) {
                        Mutex::Locker locker(_mutex);
                        _queue.push(std::move(val));
                        _cv.wakeOne();
                        return;
                }
 
                template <typename... Args> void emplace(Args &&...args) {
                        Mutex::Locker locker(_mutex);
                        _queue.emplace(std::forward<Args>(args)...);
                        _cv.wakeOne();
                        return;
                }
 
                void push(const List<T> &list) {
                        Mutex::Locker locker(_mutex);
                        for (const auto &item : list) _queue.push(item);
                        _cv.wakeAll();
                        return;
                }
 
                Error pushBlocking(const T &val, unsigned int timeoutMs = 0) {
                        Mutex::Locker locker(_mutex);
                        Error err = waitForCapacity(timeoutMs);
                        if (err != Error::Ok) return err;
                        _queue.push(val);
                        _cv.wakeOne();
                        return Error::Ok;
                }
 
                Error pushBlocking(T &&val, unsigned int timeoutMs = 0) {
                        Mutex::Locker locker(_mutex);
                        Error err = waitForCapacity(timeoutMs);
                        if (err != Error::Ok) return err;
                        _queue.push(std::move(val));
                        _cv.wakeOne();
                        return Error::Ok;
                }
 
                template <typename... Args> Error emplaceBlocking(unsigned int timeoutMs, Args &&...args) {
                        Mutex::Locker locker(_mutex);
                        Error err = waitForCapacity(timeoutMs);
                        if (err != Error::Ok) return err;
                        _queue.emplace(std::forward<Args>(args)...);
                        _cv.wakeOne();
                        return Error::Ok;
                }
 
                size_t pushDropOldest(const T &val) {
                        Mutex::Locker locker(_mutex);
                        if (_cancelled) return 0;
                        size_t dropped = 0;
                        while (_maxSize > 0 && _queue.size() >= _maxSize) {
                                _queue.pop();
                                ++dropped;
                        }
                        _queue.push(val);
                        _cv.wakeOne();
                        if (dropped > 0) _notFull.wakeAll();
                        return dropped;
                }
 
                size_t pushDropOldest(T &&val) {
                        Mutex::Locker locker(_mutex);
                        if (_cancelled) return 0;
                        size_t dropped = 0;
                        while (_maxSize > 0 && _queue.size() >= _maxSize) {
                                _queue.pop();
                                ++dropped;
                        }
                        _queue.push(std::move(val));
                        _cv.wakeOne();
                        if (dropped > 0) _notFull.wakeAll();
                        return dropped;
                }
 
                Result<T> pop(unsigned int timeoutMs = 0) {
                        Mutex::Locker locker(_mutex);
                        Error         err = _cv.wait(_mutex,
                                                     [this] { return _cancelled || !_queue.empty(); },
                                                     timeoutMs);
                        if (err != Error::Ok) return Result<T>(T{}, err);
                        if (_cancelled && _queue.empty()) return Result<T>(T{}, Error::Cancelled);
                        T ret = std::move(_queue.front());
                        _queue.pop();
                        if (_queue.empty()) _cv.wakeAll();
                        // A consumer popping always frees a slot for any
                        // producer parked on the not-full condition.
                        _notFull.wakeOne();
                        return makeResult(std::move(ret));
                }
 
                Result<T> tryPop() {
                        Mutex::Locker locker(_mutex);
                        if (_queue.empty()) return Result<T>(T{}, Error::Empty);
                        T ret = std::move(_queue.front());
                        _queue.pop();
                        if (_queue.empty()) _cv.wakeAll();
                        _notFull.wakeOne();
                        return makeResult(std::move(ret));
                }
 
                Result<T> peek(unsigned int timeoutMs = 0) {
                        Mutex::Locker locker(_mutex);
                        Error         err = _cv.wait(_mutex,
                                                     [this] { return _cancelled || !_queue.empty(); },
                                                     timeoutMs);
                        if (err != Error::Ok) return Result<T>(T{}, err);
                        if (_cancelled && _queue.empty()) return Result<T>(T{}, Error::Cancelled);
                        return makeResult(T(_queue.front()));
                }
 
                Result<T> tryPeek() {
                        Mutex::Locker locker(_mutex);
                        if (_queue.empty()) return Result<T>(T{}, Error::Empty);
                        return makeResult(T(_queue.front()));
                }
 
                Error waitForEmpty(unsigned int timeoutMs = 0) {
                        Mutex::Locker locker(_mutex);
                        return _cv.wait(_mutex, [this] { return _queue.empty() || _cancelled; }, timeoutMs);
                }
 
                bool isEmpty() const {
                        Mutex::Locker locker(_mutex);
                        return _queue.empty();
                }
 
                size_t size() const {
                        Mutex::Locker locker(_mutex);
                        return _queue.size();
                }
 
                void clear() {
                        Mutex::Locker locker(_mutex);
                        std::queue<T> empty;
                        std::swap(_queue, empty);
                        _notFull.wakeAll();
                        return;
                }
 
                void cancelWaiters() {
                        Mutex::Locker locker(_mutex);
                        _cancelled = true;
                        _cv.wakeAll();
                        _notFull.wakeAll();
                }
 
                void reset() {
                        Mutex::Locker locker(_mutex);
                        _cancelled = false;
                }
 
                bool isCancelled() const {
                        Mutex::Locker locker(_mutex);
                        return _cancelled;
                }
 
        private:
                Error waitForCapacity(unsigned int timeoutMs) {
                        if (_cancelled) return Error::Cancelled;
                        if (_maxSize == 0) return Error::Ok;
                        // Re-check after each wake: a consumer may have
                        // popped one slot and woken multiple parked
                        // producers, so any one of them must verify the
                        // cap before proceeding.
                        Error err = _notFull.wait(_mutex,
                                                  [this] { return _cancelled || _maxSize == 0 || _queue.size() < _maxSize; },
                                                  timeoutMs);
                        if (err != Error::Ok) return err;
                        if (_cancelled) return Error::Cancelled;
                        return Error::Ok;
                }
 
                mutable Mutex _mutex;
                WaitCondition _cv;
                WaitCondition _notFull;
                std::queue<T> _queue;
                size_t        _maxSize = 0;
                bool          _cancelled = false;
};
 
PROMEKI_NAMESPACE_END
 
#endif // PROMEKI_ENABLE_CORE