0
\$\begingroup\$

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.

\$\endgroup\$
1
\$\begingroup\$
  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();
    }
};
\$\endgroup\$
  • 3
    \$\begingroup\$ I agree with some of your points. I do believe, however, that void _push(E&& src) requires the addition template parameter E. This makes src a forwarding or "universal" reference, as Scott Meyers calls it. Using T&& instead indicates an r-value reference, because T has already been deduced. Scott Meyers explains this better here. \$\endgroup\$ – Frank Feb 10 '17 at 1:22

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

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