3
\$\begingroup\$

Below is my honest attempt to write thread safe queue using C++11 features. Please review it and provide your comments.

#include <queue>
#include <condition_variable>
#include <atomic>
#include <iostream>

template <typename T>
class ThreadSafeQueue
{
    std::queue<T> m_queue;
    std::mutex m_mutex;
public :

    void Push(const T& data)
    {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_queue.push(data);
    }

    // Universal reference
    template <typename ...Args>
    void Emplace(Args&&... args)
    {
        std::lock_guard<std::mutex> lock(m_mutex);
        m_queue.emplace(std::forward<Args>(args)...);  
    }

    void PopFront(T& data)
    {
        std::lock_guard<std::mutex> lock(m_mutex);
        data = m_queue.front();
        m_queue.pop();
    }

    void PopFrontNoLock(T& data)
    {
        data = m_queue.front();
        m_queue.pop();
    }

    size_t Size()
    {
        std::lock_guard<std::mutex> lock(m_mutex);
        return m_queue.size();
    }

    bool Empty()
    {
        std::lock_guard<std::mutex> lock(m_mutex);
        return m_queue.empty();
    }

    bool EmptyNoLock()
    {
        return m_queue.empty();
    }

    std::mutex& GetMutex()
    {
        return m_mutex;
    }

};

template <typename T>
class ReaderWriterQueue
{
    ThreadSafeQueue<T> m_safeQueue{};
    std::condition_variable m_condVar{};
    std::atomic<bool> m_writeCompleted{};
public :
    void Push(const T& data)
    {
        m_safeQueue.Push(data);
        m_condVar.notify_one();
    }

    // Universal reference
    template <typename ...Args>
    void Emplace(Args&&... args)
    {
        m_safeQueue.Emplace(std::forward<Args>(args)...);
        m_condVar.notify_one();
    }

    void PopFront(T& data)
    {
        std::unique_lock<std::mutex> lock(m_safeQueue.GetMutex());
        m_condVar.wait(lock, [this] {return (!m_safeQueue.EmptyNoLock() || m_writeCompleted); });
        m_safeQueue.PopFrontNoLock(data);
    }

    void NotifyWriteEnd()
    {
        m_writeCompleted = true;
    }

    bool IsWriteFinished() const
    {
        return m_writeCompleted;
    }

    bool Empty()
    {
        return m_safeQueue.Empty();
    }

};

Driver code to test thread safe queue:

void ProduceData(ReaderWriterQueue<int>& queue)
{
    int i = 0;
    while (i < 10)
    {
        queue.Emplace(i);
        i++;
    }
    queue.NotifyWriteEnd();
    std::cout << "Finished Writing" << std::endl;
}

void ReadData(ReaderWriterQueue<int>& queue)
{
    while (!queue.IsWriteFinished() || !queue.Empty())
    {
        int waitTime;
        queue.PopFront(waitTime);
        std::cout << "popped " << waitTime << "\n";
        std::this_thread::sleep_for(std::chrono::seconds(waitTime));
    }
    std::cout << "Finished Reading" << std::endl;
}


int main()
{
    ReaderWriterQueue<int> rwQueue;
    std::thread writer(ProduceData, std::ref(rwQueue));
    const int readerThreadCount = 2;
    std::thread readers[readerThreadCount];
    for (int count=0; count < readerThreadCount; count++)
    {
        std::thread reader(ReadData, std::ref(rwQueue));
        readers[count] = std::move(reader);
    }

    if (writer.joinable())
    {
        writer.join();
    }
    for (int count = 0; count < readerThreadCount; count++)
    {
        if (readers[count].joinable())
        {
            readers[count].join();
        }
    }
    return 0;
}
\$\endgroup\$

1 Answer 1

2
\$\begingroup\$

There's no way to notify locked consumers of the write end. You might want to modify PopFront and NotifyWriteEnd to return a special null value to every awaiting consumer (or otherwise wake up all the awaiting consumers and notify them the queue is exhausted) when NotifyWriteEnd() is called. Otherwise you'll need a rather serious contract inherent in your application design, that every consumer should know in advance that it would be able to get a matching producer before starting awaiting on the queue.

\$\endgroup\$
0

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.