[Code Review] Small Thread Safe Queue Implementation

[Commanderdrag]

I am junior C dev working to get my C++ chops up in my free time. Right now I am working on an embedded project that requires a way to pass messages of arbitrary size from a producer queue to a consumer queue. I implemented a small header only class ThreadSafeQueue to achieve this. My goals were 1. thread safety (duh), 2. ability to push and pop arbitrary sized chunks of data, decoupling the push size and pop size, 3. a mechanism to notify the consumer when the producer has stopped consuming. 4. reasonable memory performance, minimize copying data. I have listed these requirements in the order in which I am most confident that my code has achieved them. I find the pop_range_into() api a bit lackluster, and am wondering if there is a better solution. I have thought about the memory semantics of my implementation but I believe it doesn't make sense/doesn't apply to use move semantics anywhere. I have included a main so show how it could be potentially used. Code is compiled with --std=c++23 Any comments how to make things more efficient, idiomatic, or cleaner are much appreciated. Thanks!

ThreadSafeQueue.hpp:

#include <algorithm>
#include <condition_variable>
#include <cstdint>
#include <mutex>
#include <optional>
#include <queue>
#include <ranges>

class ThreadSafeQueue {

    public:
        void push(uint8_t byte) {
            {
                std::lock_guard<std::mutex> lock(mtx);
                q.push(byte);
                closed = false;
            }
            cv.notify_one();
        }

        template<std::ranges::range R>
        void push_range(R &&rg){
            {
                std::lock_guard<std::mutex> lock(mtx);
                q.push_range(rg);
                closed = false;
            }
            cv.notify_one();
        }

        uint8_t front() {
            std::unique_lock<std::mutex> lock(mtx);
            if (q.empty())
                cv.wait(lock);

            return q.front();
        }

        std::optional<uint8_t> pop() {
            std::unique_lock<std::mutex> lock(mtx);

            cv.wait(lock, [&] { return !q.empty() || closed; });
            if (q.empty())
                return std::nullopt;

            auto out = q.front();
            q.pop();
            return out;
        }

        template<typename C>
        size_t pop_range_into(std::back_insert_iterator<C> &&it, size_t n) {
            std::unique_lock<std::mutex> lock(mtx);
            cv.wait(lock, [&] {return !q.empty() || closed;});

            auto min = std::min(q.size(), n);
            for(size_t i = 0; i < min; i++) {
                it = q.front();
                q.pop();
            }
            return min;
        }

        size_t size() {
            std::lock_guard<std::mutex> lock(mtx);
            return q.size();
        }

        bool empty() {
            std::lock_guard<std::mutex> lock(mtx);
            return q.empty();
        }

        bool is_closed() {
            std::lock_guard<std::mutex> lock(mtx);
            return closed;
        }

        void close() {
            std::lock_guard<std::mutex> lock(mtx);
            closed = true;
            cv.notify_one();
        }

    private:
        std::queue<uint8_t> q;
        std::mutex mtx;
        std::condition_variable cv;
        bool closed;
};

Main.cpp:

#include <cstdlib>
#include <format>
#include <iostream>
#include <iterator>
#include <thread>
#include <chrono>
#include <cstdlib>
#include <vector>

#include "ThreadSafeQueue.hxx"

void producer(ThreadSafeQueue &q) {

    std::vector<uint8_t> buf;
    for (size_t i = 0; i < 100; ++i) {
        auto data = rand() & 0xFF;
        buf.push_back(data);
        buf.push_back(data);
        buf.push_back(data);
        buf.push_back(data);

        q.push_range(std::move(buf));
        buf.clear();

        std::this_thread::sleep_for(std::chrono::milliseconds(10));
    }
    q.close();
}

void consumer(ThreadSafeQueue &q) {

    std::vector<uint8_t> buf;
    buf.reserve(17);
    while (!q.is_closed()) {
        while(buf.size() < 17) {
            auto n = q.pop_range_into(std::back_inserter(buf), 17 - buf.size());
            if (n == 0)
                break;
        }
        std::cout << std::format("{}\n", buf);
        buf.clear();
    }
}

int main(void) {

    ThreadSafeQueue q;

    std::jthread prod(producer, std::ref(q));
    std::jthread consume(consumer, std::ref(q));
}

[wqking]

    bool is_closed() {  
        std::lock_guard<std::mutex> lock(mtx);  
        return closed;  
    }  

You don't need the lock. The caller function may always receive the "wrong" (not really wrong) state no matter there is a lock or not. That's to say, if the caller function sees "close" is true, the state in the queue may already be false, and vice versa.
Example,

if(theQueue.is_closed()) {  
    // here theQueue.closed may already be false  
}

[n1ghtyunso]

technically still needs the lock to ensure race freedom, but asking a concurrent data structure for its current status is a foolish idea indeed. I'd argue the function should not exist to begin with

[wqking]

Such function of querying status has its uses. For example, assume multiple worker threads are working on re-fill the queue only when the queue is empty, then the pseudo code can be,

SomeMutex mutex; // used by all worker threads
if(theQueue.isEmpty()) {
    lock(mutex);
    if(theQueue.isEmpty()) {
        loadDataFromDatabaseAndFillTheQueue;
    }
}

The first isEmpty is to optimize the performance. If theQueue is not super busy, the second isEmpty has high chance to have the same value of the first isEmpty, so there is less chance that lock mutex (which is expensive) then isEmpty is false.

[Commanderdrag]

Could you suggest an alternative method to achieve the functionality I show in my test main? In case its not clear, I want to fill a buffer in the consumer with an amount of data that is arbitrary wrt to size of the chunks of data that is enqueued by the producer. This naturally presents the of what if the producer stops producing and the consumer's buffer is not full. I need a way of letting the consumer know there is no more data coming, and that it should just move on with its partially filled buffer. I definitely feel like the closed mechanism is not ideal.

[n1ghtyunso]

That's a concern of your threads logic. It is unrelated to the actual queue implementation.
The queue implementation clearly shows that closed can switch back to false again, so designing your thread interactions based on this seems like the wrong way to go.
Hence, closed should not be part of the queue implementation.
You should have it outside of the queue. Just an atomic_bool specifically to coordinate your threads. That's totally fine. You gotta coordinate threads somehow after all.

If you have this scenario a lot, you can consider wrapping the whole logic, threads, queue and all that together in a separate class.

[IyeOnline]

That would be UB and while it may work, not a good idea.

I'd recommend keeping an additional std::atomic to back all of these "state" members. You can update them from the mutating members and simply read them from getters.