7
\$\begingroup\$

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();
}

\$\endgroup\$
6
  • \$\begingroup\$ I would not that most people distinguish between pop() and top() (AKA front()). See: en.cppreference.com/w/cpp/container/queue. pop() removes value without returning it, while top() (front()) returns the value. \$\endgroup\$ Mar 7 at 22:43
  • \$\begingroup\$ You're right, but most of the time I use front() followed by pop(), so I think it's not that bad to connect these 2 functions into 1. \$\endgroup\$
    – Edziju
    Mar 7 at 22:49
  • \$\begingroup\$ The usual reason for separating them is that a pop() that returns a value is not safe in the presence of exceptions. See: stackoverflow.com/a/25035949/14065 (I must admit I know of the rule but have not done extensive research and simply googled why they are normally seporated). Hence this is only a comment. \$\endgroup\$ Mar 7 at 23:00
  • \$\begingroup\$ Hmmm... I need to delve deeper into this topic because before implementing my own thread-safe queue, I've looked at many implementations from other people and I don't think I've ever seen separate pop and front() methods. \$\endgroup\$
    – Edziju
    Mar 7 at 23:03
  • 1
    \$\begingroup\$ I think in a multithreaded context you might want the pop() to return the value as you might not be able to front() then pop() the same value \$\endgroup\$
    – Alex
    Mar 7 at 23:05

2 Answers 2

4
\$\begingroup\$

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();
}
\$\endgroup\$
8
  • 1
    \$\begingroup\$ Real question, why would the std move be necessary in try_pop? Wouldn't RVO be applied? I'll admit I don't even see why the std move is necessary in pop. \$\endgroup\$
    – Alex
    Mar 7 at 23:34
  • 1
    \$\begingroup\$ @Alex Good point. Not all versions of C++ have guaranteed RVO, if you can rely on it then it wouldn't be necessary in try_pop(). In pop() it is different, the call to pop_opt() itself might use RVO, but then you have to get the value out of the optional. \$\endgroup\$
    – G. Sliepen
    Mar 7 at 23:45
  • 2
    \$\begingroup\$ @Edziju But now consider two threads, both doing while(!queue.empty()) { auto val = queue.pop(); }, and then you have a problem: they could both evaluate queue.empty() when there is only one item in the queue. \$\endgroup\$
    – G. Sliepen
    Mar 7 at 23:46
  • 3
    \$\begingroup\$ @Edziju But then why not just do while(!queue.QuitRequested()) { while (auto val = queue.try_pop()) {…} }? You can do it without needing queue.empty(). \$\endgroup\$
    – G. Sliepen
    Mar 7 at 23:56
  • 2
    \$\begingroup\$ +1 for "remove size() and empty()". (There's arguably a valid use case for size() in statistics reporting, but even then it would be better to name it e.g. estimate_size() to avoid creating false expectations.) \$\endgroup\$ Mar 8 at 12:50
5
\$\begingroup\$

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).

\$\endgroup\$
2
  • \$\begingroup\$ Okey, what if instead of std::atomic_bool quit I'd do std::stop_source source. request_stop() function would do source.request_stop(). and pop_opt() instead of the while loop -> cv.wait(lock, source.get_token(), [&queue](){ return !queue.empty() }); ? \$\endgroup\$
    – Edziju
    Mar 7 at 23:10
  • \$\begingroup\$ Would it be as simple as just adding a std::scoped_lock (or unique_lock) before quit = true;? I think so, unless I'm missing something. But good spot! \$\endgroup\$
    – Raf
    Mar 8 at 11:57

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge that you have read and understand our privacy policy and code of conduct.

Not the answer you're looking for? Browse other questions tagged or ask your own question.