blocking_queue.hpp

Go to the documentation of this file.

// Copyright (c) 2021 MIT Digital Currency Initiative,
//                    Federal Reserve Bank of Boston
// Distributed under the MIT software license, see the accompanying
// file COPYING or http://www.opensource.org/licenses/mit-license.php.
 
#ifndef OPENCBDC_TX_SRC_COMMON_BLOCKING_QUEUE_H_
#define OPENCBDC_TX_SRC_COMMON_BLOCKING_QUEUE_H_
 
#include <condition_variable>
#include <functional>
#include <mutex>
#include <queue>
 
namespace cbdc {
    template<typename T, typename Q>
    class blocking_queue_internal {
      public:
        blocking_queue_internal() = default;
 
        blocking_queue_internal(const blocking_queue_internal&) = delete;
        auto operator=(const blocking_queue_internal&)
            -> blocking_queue_internal& = delete;
 
        blocking_queue_internal(blocking_queue_internal&&) = delete;
        auto operator=(blocking_queue_internal&&)
            -> blocking_queue_internal& = delete;
 
        ~blocking_queue_internal() {
            clear();
        }
 
        auto push(const T& item) -> size_t {
            auto sz = [&]() {
                std::unique_lock<std::mutex> lck(m_mut);
                m_buffer.push(item);
                m_wake = true;
                return m_buffer.size();
            }();
            m_cv.notify_one();
            return sz;
        }
 
        [[nodiscard]] auto pop(T& item) -> bool {
            {
                std::unique_lock<std::mutex> lck(m_mut);
                if(m_buffer.empty()) {
                    m_cv.wait(lck, [&] {
                        return m_wake;
                    });
                }
 
                bool popped{false};
                if(!m_buffer.empty()) {
                    item = std::move(first_item<T, Q>());
                    m_buffer.pop();
                    popped = true;
                    m_wake = !m_buffer.empty();
                }
 
                return popped;
            }
        }
 
        void clear() {
            {
                std::unique_lock<std::mutex> lck(m_mut);
                m_buffer = decltype(m_buffer)();
                m_wake = true;
            }
            m_cv.notify_all();
        }
 
        void reset() {
            std::unique_lock l(m_mut);
            m_wake = false;
        }
 
      private:
        template<typename TT, typename QQ>
        auto first_item() ->
            typename std::enable_if<std::is_same<QQ, std::queue<TT>>::value,
                                    const TT&>::type {
            return m_buffer.front();
        }
 
        template<typename TT, typename QQ>
        auto first_item() ->
            typename std::enable_if<!std::is_same<QQ, std::queue<TT>>::value,
                                    const TT&>::type {
            return m_buffer.top();
        }
 
        Q m_buffer;
        std::mutex m_mut;
        std::condition_variable m_cv;
        bool m_wake{false};
    };
 
    template<typename T>
    using blocking_queue = blocking_queue_internal<T, std::queue<T>>;
 
    template<typename T, typename C = std::less<T>>
    using blocking_priority_queue
        = blocking_queue_internal<T,
                                  std::priority_queue<T, std::vector<T>, C>>;
}
 
#endif // OPENCBDC_TX_SRC_COMMON_BLOCKING_QUEUE_H_