thread_queue.h

Go to the documentation of this file.

 
/*******************************************************************************
 * Copyright (c) 2017-2022 Frank Pagliughi <fpagliughi@mindspring.com>
 *
 * All rights reserved. This program and the accompanying materials
 * are made available under the terms of the Eclipse Public License v2.0
 * and Eclipse Distribution License v1.0 which accompany this distribution.
 *
 * The Eclipse Public License is available at
 *    http://www.eclipse.org/legal/epl-v20.html
 * and the Eclipse Distribution License is available at
 *   http://www.eclipse.org/org/documents/edl-v10.php.
 *
 * Contributors:
 *    Frank Pagliughi - initial implementation and documentation
 *******************************************************************************/
 
#ifndef __mqtt_thread_queue_h
#define __mqtt_thread_queue_h
 
#include <algorithm>
#include <condition_variable>
#include <deque>
#include <limits>
#include <mutex>
#include <queue>
#include <thread>
 
namespace mqtt {

class queue_closed : public std::runtime_error
{
public:
    queue_closed() : std::runtime_error("queue is closed") {}
};


template <typename T, class Container = std::deque<T>>
class thread_queue
{
public:
    using container_type = Container;
    using value_type = T;
    using size_type = typename Container::size_type;

    static constexpr size_type MAX_CAPACITY = std::numeric_limits<size_type>::max();
 
private:
    mutable std::mutex lock_;
    std::condition_variable notEmptyCond_;
    std::condition_variable notFullCond_;
    size_type cap_{MAX_CAPACITY};
    bool closed_{false};

    std::queue<T, Container> que_;

    using guard = std::lock_guard<std::mutex>;
    using unique_guard = std::unique_lock<std::mutex>;

    bool is_done() const { return closed_ && que_.empty(); }
 
public:
    thread_queue() {}
    explicit thread_queue(size_t cap) : cap_(std::max<size_type>(cap, 1)) {}
    bool empty() const {
        guard g{lock_};
        return que_.empty();
    }

    size_type capacity() const {
        guard g{lock_};
        return cap_;
    }

    void capacity(size_type cap) {
        guard g{lock_};
        cap_ = std::max<size_type>(cap, 1);
        if (cap_ > que_.size())
            notFullCond_.notify_all();
    }

    size_type size() const {
        guard g{lock_};
        return que_.size();
    }

    void close() {
        guard g{lock_};
        closed_ = true;
        notFullCond_.notify_all();
        notEmptyCond_.notify_all();
    }

    bool closed() const {
        guard g{lock_};
        return closed_;
    }

    bool done() const {
        guard g{lock_};
        return is_done();
    }

    void clear() {
        guard g{lock_};
        while (!que_.empty()) que_.pop();
        notFullCond_.notify_all();
    }

    void put(value_type val) {
        unique_guard g{lock_};
        notFullCond_.wait(g, [this] { return que_.size() < cap_ || closed_; });
        if (closed_)
            throw queue_closed{};
 
        que_.emplace(std::move(val));
        notEmptyCond_.notify_one();
    }

    bool try_put(value_type val) {
        guard g{lock_};
        if (que_.size() >= cap_ || closed_)
            return false;
 
        que_.emplace(std::move(val));
        notEmptyCond_.notify_one();
        return true;
    }

    template <typename Rep, class Period>
    bool try_put_for(value_type val, const std::chrono::duration<Rep, Period>& relTime) {
        unique_guard g{lock_};
        bool to = !notFullCond_.wait_for(g, relTime, [this] {
            return que_.size() < cap_ || closed_;
        });
        if (to || closed_)
            return false;
 
        que_.emplace(std::move(val));
        notEmptyCond_.notify_one();
        return true;
    }

    template <class Clock, class Duration>
    bool try_put_until(
        value_type val, const std::chrono::time_point<Clock, Duration>& absTime
    ) {
        unique_guard g{lock_};
        bool to = !notFullCond_.wait_until(g, absTime, [this] {
            return que_.size() < cap_ || closed_;
        });
 
        if (to || closed_)
            return false;
 
        que_.emplace(std::move(val));
        notEmptyCond_.notify_one();
        return true;
    }

    bool get(value_type* val) {
        if (!val)
            return false;
 
        unique_guard g{lock_};
        notEmptyCond_.wait(g, [this] { return !que_.empty() || closed_; });
        if (que_.empty())  // We must be done
            return false;
 
        *val = std::move(que_.front());
        que_.pop();
        notFullCond_.notify_one();
        return true;
    }

    value_type get() {
        unique_guard g{lock_};
        notEmptyCond_.wait(g, [this] { return !que_.empty() || closed_; });
        if (que_.empty())  // We must be done
            throw queue_closed{};
 
        value_type val = std::move(que_.front());
        que_.pop();
        notFullCond_.notify_one();
        return val;
    }

    bool try_get(value_type* val) {
        if (!val)
            return false;
 
        guard g{lock_};
        if (que_.empty())
            return false;
 
        *val = std::move(que_.front());
        que_.pop();
        notFullCond_.notify_one();
        return true;
    }

    template <typename Rep, class Period>
    bool try_get_for(value_type* val, const std::chrono::duration<Rep, Period>& relTime) {
        if (!val)
            return false;
 
        unique_guard g{lock_};
        notEmptyCond_.wait_for(g, relTime, [this] { return !que_.empty() || closed_; });
 
        if (que_.empty())
            return false;
 
        *val = std::move(que_.front());
        que_.pop();
        notFullCond_.notify_one();
        return true;
    }

    template <class Clock, class Duration>
    bool try_get_until(
        value_type* val, const std::chrono::time_point<Clock, Duration>& absTime
    ) {
        if (!val)
            return false;
 
        unique_guard g{lock_};
        notEmptyCond_.wait_until(g, absTime, [this] { return !que_.empty() || closed_; });
        if (que_.empty())
            return false;
 
        *val = std::move(que_.front());
        que_.pop();
        notFullCond_.notify_one();
        return true;
    }
};

}  // namespace mqtt
 
#endif  // __mqtt_thread_queue_h