Another thread safe queue implementation

Question

I have a class, Queue, that I tried to make thread safe. It has these three member variables:

    std::queue<T>   m_queue;
    pthread_mutex_t m_mutex;
    pthread_cond_t  m_condition;

and a push and pop implemented as:

    template<class T> void Queue<T>::push(T value)
    {
        pthread_mutex_lock( &m_mutex );
        m_queue.push(value);
        if( !m_queue.empty() )
        {
            pthread_cond_signal( &m_condition );
        }
        pthread_mutex_unlock( &m_mutex );
    }

    template<class T> bool Queue<T>::pop(T& value, bool block)
    {
        bool rtn = false;
        pthread_mutex_lock( &m_mutex );
        if( block )
        {
            while( m_queue.empty() )
            {
                pthread_cond_wait( &m_condition, &m_mutex );
            }
        }
        if( !m_queue.empty() )
        {
            value = m_queue.front();
            m_queue.pop();  
            rtn = true;
        }
        pthread_mutex_unlock( &m_mutex );
        return rtn;
    }

Unfortunately there are occasional issues that may be the fault of this code. That is, there are two threads and sometimes thread 1 never comes out of push() and at other times thread 2 never comes out of pop() (the block parameter is true) though the queue isn't empty.

I understand there are other implementations available, but I'd like to try to fix this code, if needed. Anyone see any issues?

The constructor has the appropriate initializations:

    Queue()
    {
        pthread_mutex_init( &m_mutex, NULL );
        pthread_cond_init( &m_condition, NULL );
    }

and the destructor, the corresponding 'destroy' calls.

Are you running it in a debugger like `gdb`? Are `push()` or `pop()` throwing exceptions/crashing in the middle? — Anthony, Dec 19 '12 at 23:03
Sorry to say that, but your code doesn't contain any error... And it is well written... Try to debug with gdb, but the problem is not where you are looking — benjarobin, Dec 19 '12 at 23:16
Minor fix : In push function `if( !m_queue.empty() )` is always true — benjarobin, Dec 19 '12 at 23:22
You should use RAII for the the calls to lock/unlock. Otherwise an exception is going to cause problems with the locks. It may not be obvious but there are potential exception conditions (especially if T is not simple). — Martin York, Dec 19 '12 at 23:32
Your push should not conditionally signal the condition variable. You just pushed the value, and by definition, no one can fulfill their predicate evaluation without the mutex that *you* own locked on their side. I would simply lock, insert, signal, unlock, and allow the poppers the right to check their predicate after they receive the signal (which they should be doing anyway) under the locked mutex coming out of the cond-wait. Lastly, from this code `empty()` should only be be called by someone that owns the mutex, and I hope you enforce this correctly. — WhozCraig, Dec 20 '12 at 00:23
I'll take out the if( !m_queue.empty() ) in the `push` function; that's clearly not needed. And the `block` parameter is always set true, as I use pop in a while condition which I want to block. — jensph, Dec 20 '12 at 18:19
I thought of using an RAII wrapper to lock/unlock, but in the `pop` function, I wasn't sure how to handle the pthread_cond_wait() - the fact that it unlocks the mutex won't cause any problem? — jensph, Dec 20 '12 at 18:22
Have you taken a look at the the michael scott queue ? http://www.cs.rochester.edu/research/synchronization/pseudocode/queues.html You might also need this : http://stackoverflow.com/questions/151841/high-level-compare-and-swap-cas-functions — Jason Zhu, Jan 07 '13 at 16:03

score 1 · Answer 1 · edited Dec 20 '12 at 01:49

1

As mentioned by Paul Rubel, you need to initialise the mutex first. It may be a good idea at this point to wrap the mutex in a class that will handle your initialisation and finalisation for you. For example:

class mutex {
private:
    mutex(const mutex &m);
    mutex &operator=(const mutex &m);

    // OR inherit from boost::noncopyable
public:
    mutex() {
        pthread_mutex_init(&mut_, nullptr);
    }

    ~mutex() {
        pthread_mutex_destroy(&mut_);
    }

    pthread_mutex_t get() const { return mut_; }

private:
    pthread_mutex_t mut_;
};

Of note is the Boost.Threading Library which contains very well-written synchronisation classes.

edited Dec 20 '12 at 01:49

GManNickG

494,350
52
494
543

answered Dec 19 '12 at 22:46

Anthony

12,177
9
69
105

2

Also [std::mutex](http://en.cppreference.com/w/cpp/thread/mutex) class if you're fortunate to be using a C++11-compliant toolchain with a well-impelemented [thread support library](http://en.cppreference.com/w/cpp/thread). – WhozCraig Dec 19 '12 at 22:58
@WhozCraig If only we could all be so lucky! – Anthony Dec 19 '12 at 22:59
1

@anthony-arnold: For your class, you should remove the copy-cosntructor and assignment operator, or you're violating the rule of three. – GManNickG Dec 20 '12 at 01:32

score 0 · Answer 2 · answered Dec 19 '12 at 22:34

0

You must initialize the mutex before using it: pthread_mutex_init

//somewhere before push/pop
pthread_mutex_init(&m_mutex)

answered Dec 19 '12 at 22:34

Paul Rubel

26,632
7
60
80

Thanks for the help, though I am already doing so. I just wasn't as complete as I should have been in my post. – jensph Dec 19 '12 at 22:47

score 0 · Answer 3 · answered Dec 19 '12 at 23:27

0

This is not related to your problem at all, but you could fix the push function :

template<class T> void Queue<T>::push(T value)
{
    pthread_mutex_lock( &m_mutex );
    if( m_queue.empty() )
    {
        m_queue.push(value);
        pthread_cond_signal( &m_condition );
    }
    else
    {
        m_queue.push(value);
    }
    pthread_mutex_unlock( &m_mutex );
}

answered Dec 19 '12 at 23:27

benjarobin

4,410
27
21

Or just `if(m_queue.empty()) pthread_cond_signal(&m_condition); m_queue.push(value);` Since this is all inside the lock, it's okay to signal before pushing the value. – Pete Becker Dec 19 '12 at 23:35
Looks like I can take out the `if(m_queue.empty())` altogether. Thanks. – jensph Dec 20 '12 at 18:22

score 0 · Accepted Answer · answered Jan 07 '13 at 16:00

0

I realized the problems occurred when testing a build for ARM. The solution was to update the pthreads library. With the updated pthreads, everything runs fine.

answered Jan 07 '13 at 16:00

jensph

763
1
10
22

Another thread safe queue implementation

4 Answers4