safe_queue.h

Go to the documentation of this file.

// SPDX-FileCopyrightText: Copyright (c) 2024 Cisco Systems
// SPDX-License-Identifier: BSD-2-Clause
 
#pragma once
 
#include <atomic>
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <optional>
#include <queue>
 
namespace quicr {
 
    template<typename T>
    class SafeQueue
    {
      public:
        SafeQueue(uint32_t limit = 5'000)
          : stop_waiting_{ false }
          , limit_{ limit }
        {
        }
 
        ~SafeQueue() { StopWaiting(); }
 
        bool Push(T const& elem)
        {
            bool rval = true;
 
            std::lock_guard<std::mutex> _(mutex_);
 
            if (queue_.empty()) {
                cv_.notify_one();
                empty_ = false;
            }
 
            else if (limit_ != 0 && queue_.size() >= limit_) { // Make room by removing first element
                queue_.pop();
                rval = false;
            }
 
            queue_.push(elem);
 
            return rval;
        }
 
        bool Push(T&& elem)
        {
            bool rval = true;
 
            std::lock_guard<std::mutex> _(mutex_);
 
            if (queue_.empty()) {
                cv_.notify_one();
                empty_ = false;
            }
 
            else if (limit_ != 0 && queue_.size() >= limit_) { // Make room by removing first element
                queue_.pop();
                rval = false;
            }
 
            queue_.emplace(std::move(elem));
 
            return rval;
        }
 
        std::optional<T> Pop()
        {
            std::lock_guard<std::mutex> _(mutex_);
            return PopInternal();
        }
 
        std::optional<T> Front()
        {
            std::lock_guard<std::mutex> _(mutex_);
 
            if (queue_.empty()) {
                return std::nullopt;
            }
 
            return queue_.front();
        }
 
        void PopFront()
        {
            std::lock_guard<std::mutex> _(mutex_);
 
            PopFrontInternal();
        }
 
        std::optional<T> BlockPop()
        {
            std::unique_lock<std::mutex> lock(mutex_);
            cv_.wait(lock, [&]() { return (stop_waiting_ || (queue_.size() > 0)); });
 
            if (stop_waiting_) {
                return std::nullopt;
            }
 
            return PopInternal();
        }
 
        size_t Size()
        {
            std::lock_guard<std::mutex> _(mutex_);
            return queue_.size();
        }
 
        void Clear()
        {
            std::lock_guard<std::mutex> _(mutex_);
            std::queue<T> empty;
            std::swap(queue_, empty);
            empty_ = true;
        }
 
        bool Empty() const { return empty_; }
 
        void StopWaiting()
        {
            std::lock_guard<std::mutex> _(mutex_);
            stop_waiting_ = true;
            cv_.notify_all();
        }
 
        void SetLimit(uint32_t limit)
        {
            std::lock_guard<std::mutex> _(mutex_);
            limit_ = limit;
        }
 
      private:
        std::optional<T> PopInternal()
        {
            if (queue_.empty()) {
                empty_ = true;
                return std::nullopt;
            }
 
            auto elem = queue_.front();
            queue_.pop();
 
            if (queue_.empty()) {
                empty_ = true;
            }
 
            return elem;
        }
 
        void PopFrontInternal()
        {
            if (queue_.empty()) {
                empty_ = true;
                return;
            }
 
            queue_.pop();
 
            if (queue_.empty()) {
                empty_ = true;
            }
        }
 
        std::atomic<bool> empty_{ true };
        bool stop_waiting_;          // Instruct threads to stop waiting
        uint32_t limit_;             // Limit of number of messages in queue
        std::condition_variable cv_; // Signaling for thread syncronization
        std::mutex mutex_;           // read/write lock
        std::queue<T> queue_;        // Queue
    };
 
} /* namespace quicr */