Thread Safe Queue

Question

I have a thread safe queue in my library c9y. It is generally used as a task queue in the task_pool class, but in can be used for any producer / consumer problem.

queue.h

#ifndef _C9Y_QUEUE_H_
#define _C9Y_QUEUE_H_

#include <mutex>
#include <condition_variable>
#include <deque>
#include <chrono>

namespace c9y
{
    using namespace std::chrono_literals;

    //! Thread Safe Queue
    //!
    //! This is a thread safe implementation of a queue.
    //!
    //! @note There is no really safe way to copy a queue, so this
    //! queue is not copyable or asingable.
    template <typename T, class Container = std::deque<T>>
    class queue
    {
    public:
        typedef Container                           container_type;
        typedef typename Container::value_type      value_type;
        typedef typename Container::size_type       size_type;
        typedef typename Container::reference       reference;
        typedef typename Container::const_reference const_reference;

        //! Create an empty queue.
        queue() {}

        //! Initialize the queue with values.
        //!
        //! @param c the container to initialize the queue with
        explicit
        queue(const Container& c)
        : container(c) {}

        //! Initialize the queue with a reange of values.
        //!
        //! @param begin an iterator to the beginning of the range
        //! @param end an iterator to the one beond the end of the range
        template <class Iterator>
        queue(const Iterator& begin, const Iterator& end)
        : container(begin, end) {}

        //! Destructor
        ~queue()
        {
            wake();
        }

        //! Push a value onto the queue.
        //!
        //! This method will push the value onto the queue and
        //! wake up a thread that is wating in pop_wait.
        //!
        //! @param value the value to push onto the queue
        void push(const value_type& value)
        {
            auto lock = std::unique_lock<std::mutex>{mutex};
            container.push_back(value);
            cond.notify_one();
        }

        //! Pop a value of the queue.
        //!
        //! This method will try to pop a value off the queue. If no value is
        //! in the queue, it will return false.
        //!
        //! @param value the value of the pop
        //! @return true if a value was poped of the queue
        bool pop(value_type& value)
        {
            auto lock = std::unique_lock<std::mutex>{mutex};
            if (!container.empty())
            {
                value = container.front();
                container.pop_front();
                return true;
            }
            else
            {
                return false;
            }
        }

        //! Pop a value of the queue, wait if nessesary.
        //!
        //! This method will try to pop a value off the queue. If no value is
        //! in the queue, it will wait until either a value is pushed onto the
        //! queue or wake is called.
        //!
        //! @param value the value of the pop
        //! @return true if a value was poped of the queue
        //!
        //! @warning It is quite simple to build a race condition with pop_wait
        //! and wake. If you intend to reliably wake all waiting threads, use
        //! pop_wait_for with a reasonable timeout.
        bool pop_wait(value_type& value)
        {
            auto lock = std::unique_lock<std::mutex>{mutex};

            if (container.empty())
            {
                cond.wait(lock);
            }

            if (!container.empty())
            {
                value = container.front();
                container.pop_front();
                return true;
            }
            else
            {
                return false;
            }
        }

        //! Pop a value of the queue, wait for a defined duration if nessesary.
        //!
        //! This method will try to pop a value off the queue. If no value is
        //! in the queue, it will wait until either a value is pushed onto the
        //! queue or wake is called.
        //!
        //! @param value the value of the pop
        //! @param duration the duration to wait for
        //! @return true if a value was poped of the queue
        bool pop_wait_for(value_type& value, std::chrono::milliseconds duration)
        {
            auto lock = std::unique_lock<std::mutex>{mutex};

            if (container.empty())
            {
                cond.wait_for(lock, duration);
            }

            if (!container.empty())
            {
                value = container.front();
                container.pop_front();
                return true;
            }
            else
            {
                return false;
            }
        }

        //! Wake up any threads that are wating in pop_wait.
        void wake()
        {
            cond.notify_all();
        }

    private:
        std::mutex              mutex;
        std::condition_variable cond;
        Container               container;

        queue(const queue& other) = delete;
        queue& operator = (const queue& other) = delete;
    };
}

#endif

Of course there are matching unit tests:

queue_test.cpp

#include <c9y/queue.h>

#include <atomic>
#include <thread>

#include <gtest/gtest.h>
#include <c9y/thread_pool.h>

using namespace std::chrono_literals;

TEST(queue, create)
{
    auto q = c9y::queue<int>{};
}

TEST(queue, consumer_producer)
{
    auto q = c9y::queue<int>{};
    auto count = std::atomic<unsigned int>{0};

    auto prod = c9y::thread_pool{[&] () {
        for (int i = 1; i < 101; i++)
        {
            q.push(i);
        }
    }, 3};

    auto cons = c9y::thread_pool{[&] () {
        int value = 0;
        while (q.pop(value))
        {
            count++;
            std::this_thread::sleep_for(1ms);
        }
    }, 3};

    prod.join();
    cons.join();

    EXPECT_EQ(300, static_cast<unsigned int>(count));
}

TEST(queue, consumer_producer_wait)
{
    auto q = c9y::queue<int>{};
    auto count = std::atomic<unsigned int>{0};

    auto prod = c9y::thread_pool{[&] () {
        for (int i = 1; i < 101; i++)
        {
            q.push(i);
            std::this_thread::sleep_for(5ms);
        }
    }, 3};

    auto cons = c9y::thread_pool{[&] () {
        int value = 0;
        while (q.pop_wait_for(value, 100ms))
        {
            count++;
            std::this_thread::sleep_for(3ms);
        }
    }, 3};

    prod.join();
    std::this_thread::sleep_for(100ms);
    q.wake();
    cons.join();

    EXPECT_EQ(300, static_cast<unsigned int>(count));
}

Just by posting it here, I already found some issues I need to look into. This code is C++11 code but the library was recently updated to C++20; so these standards should be applied. I am interested what you think.

Answer 1

Consider using perfect forwarding to initialize the container

You have three constructors, a default one and two that initialize the container based on existing data. But the constructor of std::deque has many more forms. Picking only a few of them is inconsistent, and having to copy all of them is a lot of work. Instead, consider using perfect forwarding:

template<typename... Args>
explicit queue(Args&&... args)
: container(std::forward<Args>(args)...) {}

The destructor should not call wake()

I see you have a destructor that calls wake(), but that is very unsafe. No thread should be in pop_wait() when the destructor is called. Consider what happens after the destructor returns and those other threads have woken up: they assume they still have a live queue object! So the only time it is safe to call the destructor is if no threads are using the queue object any more, and thus there is no need to call wake() in the destructor.

For a task pool it is important that you can signal the worker threads that they should stop doing any work. That has to be done in some other way; for example by setting a flag that signals that work has to stop. The destructor of the thread pool should look like:

class thead_pool {
    queue q;
    std::vector<std::thread> threads;
    …
    ~thread_pool() {
        q.stop(); // signal queue that no more items will be pushed
        for (auto& thread: threads)
            threads.join(); // not necessary if you use std::jthread
        // after the destructor exits, q will be destroyed
    }
};

So queue should have something like:

template<…>
class queue {
public:
    …
    void stop() {
        auto lock = std::unique_lock<std::mutex>{mutex};
        stopped = true;
        wake();
    }

private:
    std::mutex mutex;
    …
    bool stopped = false;
};

Always wait using a predicate

std::condition_variable::wait() can wake up spuriously (i.e., without notify_one() or notify_all() having been called), and in more complex programs it's hard to see when a condition variable can be notified. So it's almost always better to use wait() with a predicate:

bool pop_wait(value_type& value)
{
    auto lock = std::unique_lock<std::mutex>{mutex};
    cond.wait(lock, [&]{return !container.empty() || stopped;});

    if (container.empty()) // can only happen if stopped == true
        return false;

    value = container.front();
    container.pop_front();
    return true;
}

Consider returning std::optional<T>

Instead of passing a reference and returning a bool, consider returning a std::optional<T>. This is much nicer and safer to use.

Don't pass a duration in a specific unit

Instead of passing the time to wait as std::chrono::milliseconds, make it a generic type, just like the duration argument of std::condition_variable::wait_for() itself.

No need to delete the copy constructor/assignment operator

Because std::mutex and std::condition_variable are not copyable, your class will automatically be non-copyable, and you don't have to delete the copy constructor and copy assignment operator.

Allow moving data into and out of the queue

You are copying data when you push to and pop from the queue. That can be inefficient for large types, and it also prevents you from storing non-copyable types in the queue. Consider using r-value references and/or perfect forwarding to push things into the queue, i.e. by having a push() with the same overloads as std::deque::push_back(), and an emplace() with the same parameters as std::deque::emplace_back(). Use std::move() when popping an item from the queue into a temporary variable (you don't need to move it when returning the temporary).

Notify without holding the lock

It's slightly more efficient to call notify_one() and notify_all() without having the mutex locked, otherwise there is a possibility the other thread(s) will wake, see that the mutex is still locked, then they have to do a system call to wait again until the mutex is unlocked.