waitQueues.h

Go to the documentation of this file.

#ifndef __waitQueues_h_
#define __waitQueues_h_
/*
  template functions to handle wait queues
    Copyright (C) 2018  Juergen Hannappel
 
    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
    (at your option) any later version.
 
    This program 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 General Public License for more details.
 
    You should have received a copy of the GNU General Public License
    along with this program.  If not, see <https://www.gnu.org/licenses/>.
*/
#include <deque>
#include <map>
#include <set>
#include <mutex>
#include <condition_variable>
#include <stdexcept>
#include <thread>
#include <memory>
#include <iostream>
 
namespace waitQueues {
    template <typename T> class simple {
      private:
        std::deque<std::unique_ptr<T>> queue;
        std::mutex queue_mutex;
        std::condition_variable cv;
      public:
        void enqueue(std::unique_ptr<T>& item) {
            std::unique_lock<decltype(queue_mutex)> lock(queue_mutex); //(NOLINT clang-analyzer-alpha.core.CastToStruct)
            if (!queue.empty() && queue.back() == nullptr) {
                throw std::logic_error("after adding a nullptr no further stuff may be added to a wait queue");
            }
            queue.emplace_back(std::move(item));
            cv.notify_all();
        }
        template <class ... Types> void emplace(Types ... args) {
            std::unique_ptr<T> item(new T(args...));
            enqueue(item);
        }
        void signalDone() {
            std::unique_ptr<T> end(nullptr);
            enqueue(end);
        }
        void resetDone() {
            std::unique_lock<decltype(queue_mutex)> lock(queue_mutex); //(NOLINT clang-analyzer-alpha.core.CastToStruct)
            while (!queue.empty() && queue.back() == nullptr) {
                queue.pop_back();
            }
        }
        template <class ... Types> std::unique_ptr<T> dequeue(bool mayCreateNew, Types ... args) {
            std::unique_lock<decltype(queue_mutex)> lock(queue_mutex); //(NOLINT clang-analyzer-alpha.core.CastToStruct)
            while (queue.empty()) {
                if (mayCreateNew) {
                    return std::unique_ptr<T>(new T(args...));
                }
                cv.wait(lock);
            }
            auto item = std::move(queue.front());
            if (item) { // nullptrs are end markers, multiple readers need many of them...
                queue.pop_front();
            }
            return item;
        }
        template <typename durationType> std::unique_ptr<T> dequeue(durationType timeout, bool &timedOut) {
            std::unique_lock<decltype(queue_mutex)> lock(queue_mutex);
            while (queue.empty()) {
                if (cv.wait_for(lock, timeout) == std::cv_status::timeout) {
                    timedOut = true;
                    return nullptr;
                }
            }
            std::unique_ptr<T> item(std::move(queue.front()));
            if (item) { // nullptrs are end markers, multiple readers need many of them...
                queue.pop_front();
            }
            timedOut = false;
            return item;
        }
        std::unique_ptr<T> dequeue() {
            std::unique_lock<decltype(queue_mutex)> lock(queue_mutex);
            while (queue.empty()) {
                cv.wait(lock);
            }
            auto item = std::move(queue.front());
            if (item) { // nullptrs are end markers, multiple readers need many of them...
                queue.pop_front();
            }
            return item;
        }
        decltype(queue.size()) size() const {
            return queue.size();
        }
        decltype(queue.empty()) empty() const {
            return queue.empty();
        }
        template <typename iterType> void signalDone(iterType begin,
                iterType end) {
            for (auto it = begin; it != end; ++it) {
                it->join();
            }
            signalDone();
        }
        template <typename iterType> std::thread signalDoneThread(iterType begin,
                iterType end) {
            return std::thread(&simple::signalDone<iterType>, this, begin, end);
        }
    };
 
    template <typename T, bool ascending = true> class itemComparator {
      public:
        bool operator () (const T* const& lhs, const T* const& rhs) const {
            if (lhs == nullptr) { // ensure nullptrs always go at the end of the container
                return false;
            }
            if (rhs == nullptr) {
                return true;
            }
            return ascending ? (*lhs < *rhs) : (*rhs < *lhs);
        }
    };
 
    template <typename T,
              bool ascending = true,
              typename Container = std::multiset<T*, itemComparator<T, ascending>>> class ordered {
                          private:
                          Container queue;
                          std::mutex queue_mutex;
                          std::condition_variable cv;
                          public:
    void enqueue(std::unique_ptr<T>& item) {
        std::unique_lock<decltype(queue_mutex)> lock(queue_mutex); //(NOLINT clang-analyzer-alpha.core.CastToStruct)
        if (!queue.empty() && *(queue.crbegin()) == nullptr) {
            throw std::logic_error("after adding a nullptr no further stuff may be added to a wait queue");
        }
        queue.emplace(item.get());
        item.release();
        cv.notify_all();
    }
    template <class ... Types> void emplace(Types ... args) {
        std::unique_ptr<T> item(new T(args...));
        enqueue(item);
    }
    void signalDone() {
        std::unique_ptr<T> end(nullptr);
        enqueue(end);
    }
    void resetDone() {
        std::unique_lock<decltype(queue_mutex)> lock(queue_mutex); //(NOLINT clang-analyzer-alpha.core.CastToStruct)
        while (!queue.empty() && *(queue.crbegin()) == nullptr) {
            queue.erase(queue.rbegin());
        }
    }
    template <typename durationType> std::unique_ptr<T> dequeue(durationType timeout, bool &timedOut) {
        std::unique_lock<decltype(queue_mutex)> lock(queue_mutex);
        while (queue.empty()) {
            if (cv.wait_for(lock, timeout) == std::cv_status::timeout) {
                timedOut = true;
                return nullptr;
            }
        }
        auto it = queue.begin();
        auto item = *it;
        if (item != nullptr) { // nullptrs are end markers, multiple readers need many of them...
            queue.erase(it);
        }
        return std::unique_ptr<T>(item);
    }
 
    std::unique_ptr<T> dequeue() {
        std::unique_lock<decltype(queue_mutex)> lock(queue_mutex); //(NOLINT clang-analyzer-alpha.core.CastToStruct)
        while (queue.empty()) {
            cv.wait(lock);//(NOLINT clang-analyzer-alpha.core.CastToStruct)
        }
        auto it = queue.begin();
        auto item = *it;
        if (item != nullptr) { // nullptrs are end markers, multiple readers need many of them...
            queue.erase(it);
        }
        return std::unique_ptr<T>(item);
    }
    template <typename iterType> void signalDone(iterType begin,
            iterType end) {
        for (auto it = begin; it != end; ++it) {
            it->join();
        }
        enqueue(nullptr);
    }
    template <typename iterType> std::thread signalDoneThread(iterType begin,
            iterType end) {
        return std::thread(&ordered::signalDone<iterType>, this, begin, end);
    }
    decltype(queue.size()) size() const {
        return queue.size();
    }
    decltype(queue.empty()) empty() const {
        return queue.empty();
    }
                      };
 
 
    template <typename T, typename clock_type = std::chrono::steady_clock> class timed {
        private:
        std::multimap<typename clock_type::time_point, std::unique_ptr<T>> queue;
        std::mutex queue_mutex;
        std::condition_variable cv;
        public:
    void enqueue(std::unique_ptr<T>& item, typename clock_type::time_point when) {
        std::unique_lock<decltype(queue_mutex)> lock(queue_mutex); //(NOLINT clang-analyzer-alpha.core.CastToStruct)
        if (!queue.empty() && queue.crbegin()->second == nullptr) {
            throw std::logic_error("after adding a nullptr no further stuff may be added to a wait queue");
        }
        if (item == nullptr) {
            when = clock_type::time_point::max();
        }
        queue.emplace(when, std::move(item));
        cv.notify_all();
    }
    template <typename delayType> void enqueue(std::unique_ptr<T>& item, delayType delay) {
        enqueue(item, clock_type::now() + delay);
    }
    void signalDone() {
        std::unique_ptr<T> end(nullptr);
        enqueue(end, clock_type::time_point::max());
    }
    void resetDone() {
        std::unique_lock<decltype(queue_mutex)> lock(queue_mutex); //(NOLINT clang-analyzer-alpha.core.CastToStruct)
        while (!queue.empty() && *(queue.crbegin()) == nullptr) {
            queue.erase(queue.rbegin());
        }
    }
    std::unique_ptr<T> dequeue() {
        std::unique_lock<decltype(queue_mutex)> lock(queue_mutex); //(NOLINT clang-analyzer-alpha.core.CastToStruct)
        while (true) {
            while (queue.empty()) {
                cv.wait(lock);
            }
            auto it = queue.begin();
            if (! it->second) {
                return nullptr;
            }
            if (it->first < clock_type::now()) {
                std::unique_ptr<T>item(std::move(it->second));
                queue.erase(it);
                return item;
            }
            auto status = cv.wait_until(lock, it->first);
            if (status == std::cv_status::timeout) {
                auto item = std::move(it->second);
                queue.erase(it);
                return item;
            }
        }
    }
    decltype(queue.size()) size() const {
        return queue.size();
    }
    decltype(queue.empty()) empty() const {
        return queue.empty();
    }
 
                   };
 
} // namespace waitQueues
 
#endif