Turn a non-blocking concurrent queue into a blocking concurrent queue

Question

Microsoft Visual Studio offers a non-blocking concurrent queue class, concurrency::concurrent_queue, based on Intel's TBB. I am using this as a base for a blocking single producer, single consumer concurrent queue.

I am overriding the push and try_pop methods:

#include <concurrent_queue.h>
/**
 * Single producer, single consumer blocking concurrent queue.
 */
template<typename T>
class SPSCBlockingQueue final : public concurrency::concurrent_queue<T>
{
  public:

    /**
     * Push an element to the back of the queue.
     */
    void push(const T& src)
    {
        _push(src);
    }

    /**
     * Push an element to the back of the queue.
     */
    void push(T&& src)
    {
        _push(std::move(src));
    }

    /**
     * Pop an element from the head of the queue. This method will wait
     * until there is an item in the queue that can be dequeued.
     *
     * @param dest a reference to a location to store the dequeued item
     * @return {@code true} if an item was successfully dequeued, otherwise {@code false}
     */
    bool try_pop(T& dest)
    {
        std::unique_lock<std::mutex> lock(_mtx);

        auto res = concurrency::concurrent_queue<T>::try_pop(dest);
        if (!res)
        {
            // The queue must be empty.
            hasData = false;

            // Wait until there is some data...
            _cv.wait(lock, [&] { return hasData; });

            // This should now work.
            res = concurrency::concurrent_queue<T>::try_pop(dest);
        }

        // res should be true.
        return res;
    }

    /**
     * Pop an element from the head of the queue. This method will wait
     * until there is an item in the queue that can be dequeued.
     *
     * @return the dequeued item.
     */
    T pop()
    {
        T dest;
        try_pop(dest);
        return dest;
    }

  private:

    std::mutex _mtx;
    std::condition_variable _cv;
    bool hasData = false;

    template<typename E>
    void _push(E&& src)
    {
        bool notify = false;

        std::unique_lock<std::mutex> lock(_mtx);
        if (!hasData)
        {
            notify = true;
            hasData = true;
        }

        concurrency::concurrent_queue<T>::push(std::forward<E>(src));

        lock.unlock();

        if (notify)
            _cv.notify_one();
    }
};

Here is some unit test code, using the Google test framework:

TEST(CheckTools, BlockingQueue)
{
    enum Data { D1, D2, D3, DataEnd };
    SPSCBlockingQueue<Data> q;

    auto dataElements = {
        D2, D1, D2, D3, D2, D1, D1, D3, D3, D1,
        D1, D2, D1, D3, D2, D2, D3, D2, D1
    };

    vector<Data> dataToPush = dataElements;
    vector<Data> result;

    // Consumer thread...
    auto t = thread([&]
    {
        while (true)
        {
            auto dataEle = q.pop();

            if (dataEle == DataEnd)
                break;

            result.push_back(dataEle);
        }
    });

    // Producer code...
    auto i=0;
    for (auto e : dataToPush)
    {
        if (i++ % 3 == 0)
            this_thread::sleep_for(chrono::milliseconds(250));

        q.push(e);
        i++;
    }

    q.push(DataEnd);
    t.join();

    ASSERT_EQ(result, vector<Data>{ dataElements });
}

I am interested in comments regarding efficiency and correctness.

Answer 1

1. As a matter of consistency, we don't need to separate typename declaration for this private function void _push(T&& src), and we can depend on the typename declaration on the class level.

   void _push(T&& src)
   {
       bool notify = false;
   
       std::unique_lock<std::mutex> lock(_mtx);
       if (!hasData)
       {
           notify = true;
           hasData = true;
       }
   
       concurrency::concurrent_queue<T>::push(std::forward<T>(src));
   
       lock.unlock();
   
       if (notify)
           _cv.notify_one();
   }
   

2. bool try_pop(T& dest) is not part of public interface, so we can define it as private.

3. We can depend on std::forward and modify the interfaces for push as follows:

   void push(const T&& src)
   {
       _push(std::forward<T>(src));
   }
   void push(T&& src)
   {
      _push(std::forward<T>(src));
   }
   

After adapting all these changes it should be as follows. Please note I don't have the system to compile this code from my end.

#include <concurrent_queue.h>
/**
 * Single producer, single consumer blocking concurrent queue.
 */
template<typename T>
class SPSCBlockingQueue final : public concurrency::concurrent_queue<T>
{
  public:

    void push(const T&& src)
    {
        _push(std::forward<T>(src));
    }
    void push(T&& src)
    {
        _push(std::forward<T>(src));
    }

    /**
     * Pop an element from the head of the queue. This method will wait
     * until there is an item in the queue that can be dequeued.
     *
     * @return the dequeued item.
     */
    T pop()
    {
        T dest;
        try_pop(dest);
        return dest;
    }

  private:

    std::mutex _mtx;
    std::condition_variable _cv;
    bool hasData = false;

    bool try_pop(T& dest)
    {
        std::unique_lock<std::mutex> lock(_mtx);

        auto res = concurrency::concurrent_queue<T>::try_pop(dest);
        if (!res)
        {
            // The queue must be empty.
            hasData = false;

            // Wait until there is some data...
            _cv.wait(lock, [&] { return hasData; });

            // This should now work.
            res = concurrency::concurrent_queue<T>::try_pop(dest);
        }

        // res should be true.
        return res;
    }


    void _push(T&& src)
    {
        bool notify = false;

        std::unique_lock<std::mutex> lock(_mtx);
        if (!hasData)
        {
            notify = true;
            hasData = true;
        }

        concurrency::concurrent_queue<T>::push(std::forward<T>(src));

        lock.unlock();

        if (notify)
            _cv.notify_one();
    }
};