ThreadSafe std::queue

Question

I have written a threadsafe queue and I would like you to suggest what can be improved and how to write good unit tests for this implementation.

#include <mutex>
#include <queue>
#include <atomic>
#include <condition_variable>
#include <concepts>
#include <optional>

template<typename LockT>
concept Lockable = requires(LockT lock) {
    { lock.try_lock() } -> std::convertible_to<bool>;
    { lock.lock() } -> std::convertible_to<void>;
    { lock.unlock() }-> std::convertible_to<void>;
};

template<typename T, Lockable LockT = std::mutex>
struct ThreadSafeQueue
{
    ThreadSafeQueue() = default;

    void push(std::convertible_to<T> auto&& func)
    {
        {
            std::scoped_lock lock(lockT);
            queue.push(std::forward<decltype(func)>(func));
        }
        cv.notify_one();
    }

    [[nodiscard]] T pop()
    {
        T item = std::move(pop_opt().value());
        return item;
    }

    [[nodiscard]] std::optional<T> try_pop()
    {
        auto item = pop_opt();
        return item;
    }

    [[nodiscard]] inline std::size_t size()
    {
        std::scoped_lock lock(lockT);
        return queue.size();
    }

    [[nodiscard]] inline bool empty()
    {
        std::scoped_lock lock(lockT);
        return queue.empty();
    }

    inline void request_quit()
    {
        quit = true;
        cv.notify_all();
    }

    [[nodiscard]] inline bool quit_requested() const
    {
        return quit;
    }

private:
    [[nodiscard]] std::optional<T> pop_opt()
    {
        std::unique_lock lock(lockT);
        while (queue.empty() && !quit) {
            cv.wait(lock);
        }

        if (queue.empty()) {
            return std::nullopt;
        }

        auto item = std::move(queue.front());
        queue.pop();
        lock.unlock();

        return item;
    }

    std::queue<T> queue;
    LockT lockT;
    std::condition_variable_any cv;
    std::atomic_bool quit = false;
};

#include <gtest/gtest.h>
#include <vector>
#include "ThreadSafeQueue.h"

TEST(ThreadSafeQueueTest, MultipleThreads)
{
    ThreadSafeQueue<int> queue;
    constexpr static std::size_t kNumThreads = 10;
    constexpr static int kNumIterations = 1000;
    std::vector<std::thread> threads;
    threads.reserve(kNumThreads);

    for (int i = 0; i < kNumThreads; ++i) {
        threads.emplace_back([&queue]() {
            for (int j = 0; j < kNumIterations; ++j) {
                queue.push(j);
            }
        });
    }

    for (int i = 0; i < kNumThreads; ++i) {
        threads.emplace_back([&queue]() {
            for (int j = 0; j < kNumIterations; ++j) {
                int value = queue.pop();
                EXPECT_GE(value, 0);
                EXPECT_LT(value, kNumIterations);
            }
        });
    }

    for (auto& thread : threads) {
        thread.join();
    }

    EXPECT_TRUE(queue.empty());
}

int main(int argc, char** argv)
{
    ::testing::InitGoogleTest(&argc, argv);
    return RUN_ALL_TESTS();
}

Answer 1

Why is there a template parameter LockT?

I don't understand why there is a template parameter for the type of lockT, but not for the type of queue. I think I'd rather have the type of queue be a template parameter, so that you could create one with a non-default allocator. Changing the type of lock is potentially dangerous; can your ThreadSafeQueue really guarantee thread-safety if it doesn't know that kind of lock is used?

I also don't see any use case for any of the mutex types from the standard library, other than the standard one. You are only using std::scoped_lock, so you will never benefit from std::shared_mutex or any of the timed mutexes. A std::recursive_mutex also doesn't make sense, since your class never recurses itself, and mutexT is private so nothing else could take that lock.

Remove size() and empty()

I see these member functions often added to thread safe queue implementations. It might seem natural since it's like a container, and the standard containers also provide these functions. However, by the time size() and empty() return, the value they return might no longer be valid. So the caller should not trust the results, and is in fact better off not calling them at all.

Missing std::move() in try_pop()

In try_pop() you don't use std::move(), resulting in a potentially suboptimal copy being made.

I recommend that you don't use temporary variables at all if they are unnecessary; this avoids needing to use std::move() to begin with:

[[nodiscard]] T pop()
{
    return pop_opt().value();
}

[[nodiscard]] std::optional<T> try_pop()
{
    return pop_opt();
}

Answer 2

Think I see a race condition:

     thread 1:                         thread 2:
                                         pop_opt()

                                           std::unique_lock lock(lockT);
                                           while (queue.empty() && !quit) {
                                             // Thread de-sheduled.

       request_quit()
           quit = true;
           cv.notify_all();

                                              
                                             // Thread re-sheduled. 
                                             cv.wait(lock);
                                             // Thread stuck in wait
                                             // forever as it will never
                                             // receive a notify.
                                           }

Yes your quit is atomic, but the all the interactions between quit and cv also need to be atomic (i.e. inside a lock).