Implementing a Thread-Safe Queue using Condition Variables (Updated)
Tuesday, 16 September 2008
One problem that comes up time and again with multi-threaded code is how to transfer data from one thread to another. For example, one common way to parallelize a serial algorithm is to split it into independent chunks and make a pipeline — each stage in the pipeline can be run on a separate thread, and each stage adds the data to the input queue for the next stage when it's done. For this to work properly, the input queue needs to be written so that data can safely be added by one thread and removed by another thread without corrupting the data structure.
Basic Thread Safety with a Mutex
The simplest way of doing this is just to put wrap a non-thread-safe queue, and protect it with a mutex (the examples use the types and functions from the upcoming 1.35 release of Boost):
template<typename Data>
class concurrent_queue
{
private:
std::queue<Data> the_queue;
mutable boost::mutex the_mutex;
public:
void push(const Data& data)
{
boost::mutex::scoped_lock lock(the_mutex);
the_queue.push(data);
}
bool empty() const
{
boost::mutex::scoped_lock lock(the_mutex);
return the_queue.empty();
}
Data& front()
{
boost::mutex::scoped_lock lock(the_mutex);
return the_queue.front();
}
Data const& front() const
{
boost::mutex::scoped_lock lock(the_mutex);
return the_queue.front();
}
void pop()
{
boost::mutex::scoped_lock lock(the_mutex);
the_queue.pop();
}
};
This design is subject to race conditions between calls to empty, front and pop if there
is more than one thread removing items from the queue, but in a single-consumer system (as being discussed here), this is not a
problem. There is, however, a downside to such a simple implementation: if your pipeline stages are running on separate threads,
they likely have nothing to do if the queue is empty, so they end up with a wait loop:
while(some_queue.empty())
{
boost::this_thread::sleep(boost::posix_time::milliseconds(50));
}
Though the sleep avoids the high CPU consumption of a direct busy wait, there are still some obvious downsides to
this formulation. Firstly, the thread has to wake every 50ms or so (or whatever the sleep period is) in order to lock the mutex,
check the queue, and unlock the mutex, forcing a context switch. Secondly, the sleep period imposes a limit on how fast the thread
can respond to data being added to the queue — if the data is added just before the call to sleep, the thread
will wait at least 50ms before checking for data. On average, the thread will only respond to data after about half the sleep time
(25ms here).
Waiting with a Condition Variable
As an alternative to continuously polling the state of the queue, the sleep in the wait loop can be replaced with a condition
variable wait. If the condition variable is notified in push when data is added to an empty queue, then the waiting
thread will wake. This requires access to the mutex used to protect the queue, so needs to be implemented as a member function of
concurrent_queue:
template<typename Data>
class concurrent_queue
{
private:
boost::condition_variable the_condition_variable;
public:
void wait_for_data()
{
boost::mutex::scoped_lock lock(the_mutex);
while(the_queue.empty())
{
the_condition_variable.wait(lock);
}
}
void push(Data const& data)
{
boost::mutex::scoped_lock lock(the_mutex);
bool const was_empty=the_queue.empty();
the_queue.push(data);
if(was_empty)
{
the_condition_variable.notify_one();
}
}
// rest as before
};
There are three important things to note here. Firstly, the lock variable is passed as a parameter to
wait — this allows the condition variable implementation to atomically unlock the mutex and add the thread to the
wait queue, so that another thread can update the protected data whilst the first thread waits.
Secondly, the condition variable wait is still inside a while loop — condition variables can be subject to
spurious wake-ups, so it is important to check the actual condition being waited for when the call to wait
returns.
Be careful when you notify
Thirdly, the call to notify_one comes after the data is pushed on the internal queue. This avoids the
waiting thread being notified if the call to the_queue.push throws an exception. As written, the call to
notify_one is still within the protected region, which is potentially sub-optimal: the waiting thread might wake up
immediately it is notified, and before the mutex is unlocked, in which case it will have to block when the mutex is reacquired on
the exit from wait. By rewriting the function so that the notification comes after the mutex is unlocked, the
waiting thread will be able to acquire the mutex without blocking:
template<typename Data>
class concurrent_queue
{
public:
void push(Data const& data)
{
boost::mutex::scoped_lock lock(the_mutex);
bool const was_empty=the_queue.empty();
the_queue.push(data);
lock.unlock(); // unlock the mutex
if(was_empty)
{
the_condition_variable.notify_one();
}
}
// rest as before
};
Reducing the locking overhead
Though the use of a condition variable has improved the pushing and waiting side of the interface, the interface for the consumer
thread still has to perform excessive locking: wait_for_data, front and pop all lock the
mutex, yet they will be called in quick succession by the consumer thread.
By changing the consumer interface to a single wait_and_pop function, the extra lock/unlock calls can be avoided:
template<typename Data>
class concurrent_queue
{
public:
void wait_and_pop(Data& popped_value)
{
boost::mutex::scoped_lock lock(the_mutex);
while(the_queue.empty())
{
the_condition_variable.wait(lock);
}
popped_value=the_queue.front();
the_queue.pop();
}
// rest as before
};
Using a reference parameter to receive the result is used to transfer ownership out of the queue in order to avoid the exception
safety issues of returning data by-value: if the copy constructor of a by-value return throws, then the data has been removed from
the queue, but is lost, whereas with this approach, the potentially problematic copy is performed prior to modifying the queue (see
Herb Sutter's Guru Of The Week #8 for a discussion of the issues). This does, of
course, require that an instance Data can be created by the calling code in order to receive the result, which is not
always the case. In those cases, it might be worth using something like boost::optional to avoid this requirement.
Handling multiple consumers
As well as removing the locking overhead, the combined wait_and_pop function has another benefit — it
automatically allows for multiple consumers. Whereas the fine-grained nature of the separate functions makes them subject to race
conditions without external locking (one reason why the authors of the SGI
STL advocate against making things like std::vector thread-safe — you need external locking to do many common
operations, which makes the internal locking just a waste of resources), the combined function safely handles concurrent calls.
If multiple threads are popping entries from a full queue, then they just get serialized inside wait_and_pop, and
everything works fine. If the queue is empty, then each thread in turn will block waiting on the condition variable. When a new
entry is added to the queue, one of the threads will wake and take the value, whilst the others keep blocking. If more than one
thread wakes (e.g. with a spurious wake-up), or a new thread calls wait_and_pop concurrently, the while
loop ensures that only one thread will do the pop, and
the others will wait.
Update: As commenter David notes below, using multiple consumers does have one problem: if there are several
threads waiting when data is added, only one is woken. Though this is exactly what you want if only one item is pushed onto the
queue, if multiple items are pushed then it would be desirable if more than one thread could wake. There are two solutions to this:
use notify_all() instead of notify_one() when waking threads, or to call notify_one()
whenever any data is added to the queue, even if the queue is not currently empty. If all threads are notified then the extra
threads will see it as a spurious wake and resume waiting if there isn't enough data for them. If we notify with every
push() then only the right number of threads are woken. This is my preferred option: condition variable notify calls
are pretty light-weight when there are no threads waiting. The revised code looks like this:
template<typename Data>
class concurrent_queue
{
public:
void push(Data const& data)
{
boost::mutex::scoped_lock lock(the_mutex);
the_queue.push(data);
lock.unlock();
the_condition_variable.notify_one();
}
// rest as before
};
There is one benefit that the separate functions give over the combined one — the ability to check for an empty queue, and
do something else if the queue is empty. empty itself still works in the presence of multiple consumers, but the value
that it returns is transitory — there is no guarantee that it will still apply by the time a thread calls
wait_and_pop, whether it was true or false. For this reason it is worth adding an additional
function: try_pop, which returns true if there was a value to retrieve (in which case it retrieves it), or
false to indicate that the queue was empty.
template<typename Data>
class concurrent_queue
{
public:
bool try_pop(Data& popped_value)
{
boost::mutex::scoped_lock lock(the_mutex);
if(the_queue.empty())
{
return false;
}
popped_value=the_queue.front();
the_queue.pop();
return true;
}
// rest as before
};
By removing the separate front and pop functions, our simple naive implementation has now become a
usable multiple producer, multiple consumer concurrent queue.
The Final Code
Here is the final code for a simple thread-safe multiple producer, multiple consumer queue:
template<typename Data>
class concurrent_queue
{
private:
std::queue<Data> the_queue;
mutable boost::mutex the_mutex;
boost::condition_variable the_condition_variable;
public:
void push(Data const& data)
{
boost::mutex::scoped_lock lock(the_mutex);
the_queue.push(data);
lock.unlock();
the_condition_variable.notify_one();
}
bool empty() const
{
boost::mutex::scoped_lock lock(the_mutex);
return the_queue.empty();
}
bool try_pop(Data& popped_value)
{
boost::mutex::scoped_lock lock(the_mutex);
if(the_queue.empty())
{
return false;
}
popped_value=the_queue.front();
the_queue.pop();
return true;
}
void wait_and_pop(Data& popped_value)
{
boost::mutex::scoped_lock lock(the_mutex);
while(the_queue.empty())
{
the_condition_variable.wait(lock);
}
popped_value=the_queue.front();
the_queue.pop();
}
};
Posted by Anthony Williams
[/ threading /] permanent link
Tags: threading, thread safe, queue, condition variable
Stumble It! 
| Submit to Reddit 
| Submit to DZone 
If you liked this post, why not subscribe to the RSS feed 
or Follow me on Twitter?
24 Comments
I had a go a implementing one of these with pthreads, not as neat as yours! <code> #ifndef __SYNCQUEUE_H #define __SYNCQUEUE_H
#include <stdio.h> #include <pthread.h>
template <class T> class ListNode { public: T item; ListNode<T> *next; };
template <class T> class SyncQueue { public: SyncQueue() { head = NULL; tail = NULL; size = 0; pthread_mutex_init(&mutex, NULL); pthread_cond_init(&cond, NULL); }
bool enqueue(const T &item) { if (pthread_mutex_lock(&mutex) != 0) { perror("Error! Couldn't lock mutex."); return false; }
ListNode<T> *node = new ListNode<T>; node->item = item;
if (size == 0) { head = tail = node; } else { tail->next = node; tail = tail->next; } size++;
if (pthread_cond_broadcast(&cond) != 0) { perror("cond broadcast error."); return false; } if (pthread_mutex_unlock(&mutex) != 0) { perror("couldn't unlock mutex."); return false; } return true; }
bool dequeue(T &ret_item) { pthread_mutex_lock(&mutex);
while (size < 1) { pthread_cond_wait(&cond, &mutex); }
if (size == 0) return false;
//printf("Queue size: %d\n", size); ret_item = head->item; ListNode<T> *t_node = head->next; delete head; head = t_node; size--;
pthread_mutex_unlock(&mutex); return true; }
int size;
private: ListNode<T> *head; ListNode<T> *tail; pthread_mutex_t mutex; pthread_cond_t cond; }; #endif </code>
Great article.
One behavior I can't understand is the following. Assume another thread is producing data and putting it on q, and the code below is the consumer:
concurrent_queue<Data> q; ... Data d; while (1) { q.wait_and_pop(d); do_something_with(d); }
runs significantly slower (for producer that produces a certain number of Data items) than:
concurrent_queue<Data> q; ... Data d; while (1) { while(q.empty()) { boost::this_thread::sleep(boost::posix_time::milliseconds(50)); }
q.wait_and_pop(d); do_something_with(d); }
How can this be?
Frank
Unfortunately, your design as well as implementation is faulty - crash guaranteed if you actually test it.
Implementation is wrong You are locking the mutex recursively (in empty() check).
Design is wrong because for all intents and purposes, this is a serial queue - all access takes an exclusive lock, so parallelization only achieves serial waiting for all users of the queue.@ohell:
Check again, there is no recursive locking, and yes I did test it. You are right that it serializes all users, but that's the best you can do with one mutex. You can do better with two mutexes, as you can allow a simultaneous push and pop.
@Frank:
What system are you testing this on? Also, what are you measuring when you say "runs slower"? Overall execution time?
As @ohell points out, this queue essentially serializes the pushes and the pops. If the mutex is highly contended, this may slow things down. If the popping thread sleeps for a bit, the pushing thread might get more than one item pushed without having to fight for the mutex, and improve the overall performance. However, it will likely increase the latency between the push and the pop.
@ohell:
I've tested as well, and it works as advertised.
@anthony:
Sorry, should've been more clear: overall execution time is what I meant. I, too, was thinking of an explanation along the lines of mutex contention, but I still can't see why the popping thread waiting would help in that case. Since wait() releases the mutex, from the pusher's perspective there is no difference between the popper being in sleep() or wait(), right?
That said, I'm working up some more tests that are outside my application to see if I can isolate the problem a little better. Wouldn't be the first time the problem was somewhere else in the code... perhaps adding the sleep() is having an unusual side effect somewhere else in the app. I'll let you know what I find.
Frank
Hi,
I'm relatively new to the whole multithreading business and currently looking for an implementation of a producer-consumer architecture... I think.
I wanted to ask, could I just copy the code presented here and maybe use it like that? Would there be any licensing issue?
Cheers!
Hi David,
Yes, you can just copy the code presented here and use it for whatever you like. There won't be any licensing issues. I'm glad you find it helpful.
Cool, thanks!
Okay, I think I have found an issue when multiple consumers use the queue. Assume all consumers are waiting for new data to be pushed onto the queue. When the producer then pushes multiple items in short succession, i.e. so quick that the first consumer to wake up cannot empty the queue again, then the_condition_variable.notify_one() is only called once (because it is blocked by the 'was_empty if' later). It seems to work for me if I replace notify_one() by notify_all().
Btw, I hope that all consumer threads waking up at the same is not a problem, but as far as I understand the notification mechanism, only one of them will acquire control over the mutex...
Let me know what you think...Hi David,
You're right. Thanks for spotting that. I guess my testing was not exhaustive enough :-(
The only impact of waking all the consumers is that they consume CPU time: if there's nothing in the queue they just treat it as a spurious wake and go back to sleep.
push() unlocks the mutex before notifying the condition variable. This gives an opportunity for another push() thread to grab the mutex before notifying the pop() thread.
If you have several threads pushing, and these push threads have lower priority than the pop() thread(s) - it would seem you (could) have a priority inversion - at the extreme, the pop() thread would never wake.
My previous comment is based on using a real-time priority-preemtive scheduler (no round-robin variant) - I forgot to state that.
Hi Ray,
By unlocking the mutex before notifying the condition variable, we do indeed allow another thread to acquire the mutex in order to push a new value on the queue. When the popping thread wakes it will block on the mutex until the new push() thread unlocks the mutex again.
If the popping thread is high priority and the pushing thread low priority, then you could have a temporary priority inversion, but that's a natural consequence of using a single mutex for push and pop. However, once the popping thread has blocked on the mutex, the scheduler will wake it as soon as the push thread unlocks the mutex, so it will only ever have to wait for one push().
On the flip side, if the popping thread is waiting on the condition variable, by unlocking the mutex before we call notify the (high priority) popping thread can wake and acquire the mutex immediately upon the call to notify, rather than having to wake and then go back to sleep because it still can't acquire the mutex.
Hi,
Great article! But it would be really helpful if you could show a pthreads version (i.e. no dependency on boost) ... I think more people are familiar with pthreads than with boost/thread
Thanks
Hi, I've found this code really useful, but am confused about the wait_and_pop() procedure.
In it, you return a reference to the first element, using: popped_value=the_queue.front();
But this is immediately followed by the_queue.pop();
When I'm using this code, I use it like so:
Data d; myQueue.wait_and_pop(d); cout << d.a_string;
But when I do this, I get a crash in the destructor of my Data instance at the time of the pop(). Why is pop() drying to call my destructor? And if that's what the expected behavior is, then how am I supposed to actually use the object stored in the queue? The moment I release it, it's deallocated.
Then again, my C++ is very rusty, and I might be making an incorrect assumption about the behavior...
Thanks!
Hi jimt,
Thanks for your comment. wait_and_pop() does not return a reference to the first element.
popped_value=the_queue.front() *copies* the variable referenced by the_queue.front() to variable referenced by popped_value using the copy-assignment operator. The references remain pointing to the same elements. This is immediately followed by the_queue.pop(), since this is required to remove the element from the queue.
pop() calls the destructor to destroy the element in the queue. That's fine, because you have a *copy* of that element in your variable d. If this is crashing, it is because you have a bug in your copy-assignment operator or your destructor. For example, have you got a pointer to dynamic storage which is being copied as a raw pointer without reference counting, and so being double-deleted?
Hi, good idea on template thread-safe cross-platfrm queue, which I need to implement for a project. This article (link) does a Windows specific void* queue, and discusses important issues:
http://www.codeproject.com/KB/threads/semaphores.aspx
I don't see use of a semaphore in the waiting mechanism, and then I see various notes about problems. I suspect this needs to be written in terms of a semiphore. Here is what that article says about that:
** There's no substitute for a Semaphore If you think you have invented a clever, faster, more efficient, easier, or whatever way of doing a semaphore without actually using a Semaphore, the chances approach unity that you have simply fooled yourself. Read Dijkstra's earlier papers where he was developing the notion of synchronization primitives that were preemptive-threading safe, and there was no InterlockedIncrement operation to help him. These are complex papers; the techniques are subtle. Only if you fully understand the issues of synchronization should you even consider trying something like this. The rest of the time, particularly if you are new to parallelism and synchronization, take this as a rule: you haven't a clue as to how to create a semaphore effect without using semaphores. I've been doing this professionally for a quarter century and I don't feel confident trying to fake out a semaphore's functionality with some other mechanism. Trust me in this: You Don't Want To Go There. **
I'm thinking of taking your basic idea, but putting in terms of semaphore like that article. I don't think there will be too many changes.
OK, the action of the "condition_variable" and the mutex implements some sort of equivalent of the semaphore. Of course mutex is a special kind of semaphore, usually used for a slightly different purpose, and likewise the wake action of the "condition_variable" is also like a binary (one count) semaphore. Though it may have spurious wakes, it seems to avoid the problem of tracking counts in the article I pointed to, that of the semaphore count tracking the queue count. Here only the queue keeps track of the count, and mutex blocks to only one section of code accessing the queue at a time. Spurious wakes only occur if a parallel thread happens to ask for queue item before the condition_variable sleeping thread can get to it, a slim timing event between producing and consuming.
Oh, a note on the crash issue noted by someone: One must also be careful about thread safeness of the items stored in the queue. Copy of an allocated and managed item (like string class) must be thread safe. For example a possible string class might delete string allocation after last holder was deleted, but keep a count of how many holders contained the actual string memory pointer. The string class is the item copied, not the string memory itself. If such class used pointers as this, the temporary condition is the count of pointers goes to 2 until the other holder is deleted. But imagine another thread also doing such copy and delete operations, having thus 3 and 4 pointers to the same actual memory of the string, across 2 threads. Only thread-safe string class using its own mutex can manage this without getting mixed up.
Could you send me example of usage your concurrent_queue? 1. One thread put elements to queue 2. Second thread get and process elements from queue (is not empty) 3. End of program when queue is empty and some flag was setup
Something like that with concurrent_queue: http://www.codeguru.com/forum/showpost.php? p=1317302&postcount=17
Best regards, MariuszCould you explain why you lock in the functions: empty(), try_pop(Data& popped_value) and wait_and_pop(Data& popped_value)? I only understand why you lock in the push(Data const& data) function because there you also unlock. Who will unlock the lock if i use for example the empty() function? If i use the empty() function and then calls the push() function, wont that mean that the lock is still held by the empty function which prevents me from pushing an item on to the queue.
I would like to have an explanation, i want to learn.
Regards, Greg
Hi Greg,
The locks are there to protect the shared data. Without the locks it is not safe to access the internal queue.
The lock is automatically released when the boost::mutex::scoped_lock object is destroyed at the closing brace of the enclosing block. This is an example of the RAII idiom.
Correction, there's no condition_variable defined in my version of boost (1.34.1) but they've added it later. If you include to condition.hpp, it includes condition_variable.hpp if you have boost 1.38
Hi Mark,
True, there's no boost::condition_variable in boost 1.34 --- I added it in boost 1.35.