Concurrent queue with pthread implementation

Question

I'm writing a multi-threaded queue implemented with pthreads. Since I came up with the code based on several tutorials from Internet, I would like to make sure there are no logic errors in my code:

template <typename T>
class ThreadQueue {
public:
    ThreadQueue() {
        pthread_mutex_init(&m_qmtx, NULL);
        pthread_cond_init(&m_condv, NULL);
    }

    ~ThreadQueue() {
        pthread_mutex_lock(&m_qmtx);
        m_queue.clear();
        pthread_mutex_unlock(&m_qmtx);
    }

    void push(T t_data) {
        pthread_mutex_lock(&m_qmtx);
        m_queue.push_back(t_data);
        pthread_mutex_unlock(&m_qmtx);

        pthread_cond_signal(&m_condv);
    }

    T front() {
        T ret;
        pthread_mutex_lock(&m_qmtx);
        while (m_queue.empty()) {
            pthread_cond_wait(&m_condv, &m_qmtx);
        }
        ret = m_queue.front();
        pthread_mutex_unlock(&m_qmtx);
        return ret;
    }

    void pop() {
        pthread_mutex_lock(&m_qmtx);
        if (!m_queue.empty())
            m_queue.pop_front();
        pthread_mutex_unlock(&m_qmtx);
    }

private:
    std::deque<T> m_queue;
    pthread_mutex_t m_qmtx;
    pthread_cond_t m_condv;
};

Answer 1

Correctly initialized members.

ThreadQueue() {
    pthread_mutex_init(&m_qmtx, NULL);
    pthread_cond_init(&m_condv, NULL);
}

But you don't check the error codes.
The object is potentially created with invalid members. Use exceptions to gurantee the state of your object.

Destructor does not do anything useful.

~ThreadQueue() {
    pthread_mutex_lock(&m_qmtx);
    m_queue.clear();
    pthread_mutex_unlock(&m_qmtx);
}

The destructr is supposed to clean resources.
It does not. You should destroy the mutex and condition variable in the destructor (reverse order of creation to be consistent).

Also locking before clearing the queue does not buy you anything. If at the point the destructor is called the object is still visible to other threads then it is going to break anyway. The queue is going to call clear in its own destructor so it is pointless doing it manually.

The call to signal should be inside the lock.

void push(T t_data) {
    pthread_mutex_lock(&m_qmtx);
    m_queue.push_back(t_data);
    pthread_mutex_unlock(&m_qmtx);

    pthread_cond_signal(&m_condv);
}

Since you should be using RAII for you locking anyway (See below) the signal is going to fall inside the lock stratergy unless you do something complex.

Also you should pass the parameter by const reference rather than value.

Normally front() is going to give you a reference to the front object.

T front() {
    T ret;
    pthread_mutex_lock(&m_qmtx);
    while (m_queue.empty()) {
        pthread_cond_wait(&m_condv, &m_qmtx);
    }
    ret = m_queue.front();
    pthread_mutex_unlock(&m_qmtx);
    return ret;
}

Returning a copy is probably not what you want (especially for anything interesting). But returning a reference is not really an option either as that opens you to situations where you have race conditions (you have a reference to the top object and another thread pops it (destroying it) just before you call a method).

So you need to provide a method for accessing the object while maintaining the lock (this will probably require the return of a wrapper object that maintains a lock on the queue) or alternatively removing this function.

Other non C++ things that should be addressed:

The mutex should be locked/unlocked using RAII. You are leaving your code open to exception handling problems. If an exception is generated by T which you have no control over then you could leave your queue in an unusable state with the mutex locked and no way to unlock it (because the stack of the locker has been unwound past the unlock because of an exception).

Yes this is correct:

   if (!m_queue.empty())
        m_queue.pop_front();

But don't get into bad habbits. No using the braces can lead to maintenance problems on C/C++ code because of macros. Best practice dictates that you always use {} for sub statements of if/while/for.

To make the code easy to read when including the code in the interface declaration put the variables at the top. Thus you can read the code in with the context of the variables and knowing their type. If you interface declaration is just an interface (and the code is put in the source file rather than the header) then it is fine to put the private variables at the bottom.

Answer 2

Personally, I think you've wrapped too many different responsibilities into the queue class. I think I'd start with a small wrapper for a mutex:

class mutex { 
    pthread_mutex_t m;
public:
    mutex()       { pthread_mutex_init(&m);    }
    ~mutex()      { pthread_mutex_destroy(&m); }
    void lock()   { pthread_mutex_lock(&m);    }
    void unlock() { pthread_mutex_unlock(&m);  }
};

Then I'd write something similar for your condvar. Again, you should really call pthread_cond_destroy when you're done with the condvar. A wrapper that automatically called pthread_cond_init in its ctor and pthread_cond_destroy in its dtor makes it much easier to ensure the code works correctly.

Then I'd write a small RAII wrapper for locking and unlocking the mutex:

class lock {
    mutex &m;
public:
    lock(mutex &m) : m(m) { m.lock(); }
    ~lock() { m.unlock(); }
};

Then, getting the rest of your code correct becomes rather simpler -- creating and destroying the mutex and condvar happen automatically (fixing a couple of bugs, since your code doesn't currently destroy either as it should). Code using the mutex (for example) becomes simpler as well. For example, your front() simplifies (a little) to something like this:

T front() { 
    T ret;
    lock(m_qmtx);
    while (m_queue.empty())
        pthread_cond_wait(&m_condv, &m_qmtx);
    ret = m_queue.front();
    return ret;
}

I see (at least) two obvious advantages to this:

1. Simpler code that's easier to ensure is correct.
2. The code is much closer to what you'd likely write in C++11, so anybody familiar with C++11 will find it easier to follow.

One other detail: I think I'd try t come up with better names than m_condv and m_qmtx. At least in my opinion, your current names are next to meaningless (and ugly to boot).

Answer 3

The pthread_cond_signal should be done with the mutex locked. And a broadcast might be preferable to a signal.

I'd put the private variables at the start of the class and braces on the if statements, even though they are of only one line.

Answer 4

Your use of the mutex and condition variable seem correct.

It's a queue. Use a std::queue internally and mimic its interface. You're missing a lot of things. Here's the interface of one I wrote, which exactly mimics std::queue...

template<typename T, typename Container = std::deque<T>>
class ConcurrentQueue
{
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;

                                 explicit ConcurrentQueue(const Container& c);
                                 explicit ConcurrentQueue(Container&& c = Container());
                                 ConcurrentQueue(const ConcurrentQueue& other);
                                 ConcurrentQueue(ConcurrentQueue&& other);
    template<typename Allocator> explicit ConcurrentQueue(const Allocator& a);
    template<typename Allocator> ConcurrentQueue(const Container& c, const Allocator& a);
    template<typename Allocator> ConcurrentQueue(Container&& c, const Allocator& a);
    template<typename Allocator> ConcurrentQueue(const ConcurrentQueue& other, const Allocator& a);
    template<typename Allocator> ConcurrentQueue(ConcurrentQueue&& other, const Allocator& a);

    ConcurrentQueue& operator=(const ConcurrentQueue& other);
    ConcurrentQueue& operator=(ConcurrentQueue&& other);

    bool empty() const;
    size_type size() const;

    void push(const T& item);
    void push(T&& item);
    template< class... Args > 
    void emplace(Args&&... args );
    T pop();
    void swap(ConcurrentQueue& other);

private:
    std::queue<T, Container> m_Queue;
    /* ... */
};

I'll leave it to you to fill in the implementation. I used the C++ standard library, not pthread. I recommend you do too. Note I also called it ConcurrentQueue. It's a more standard name than ThreadQueue for this class. SharedQueue is also quite popular (needless to say, this is a popular thing to write).